131 lines
4.4 KiB
Python
Executable File
131 lines
4.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""港股新股检查脚本 - 集思录"""
|
|
|
|
from playwright.sync_api import sync_playwright
|
|
from datetime import datetime
|
|
import json
|
|
|
|
CONFIG_FILE = "/root/.openclaw/workspace/hk_ipo_state.json"
|
|
REPORT_FILE = "/root/.openclaw/workspace/hk_ipo_report.txt"
|
|
|
|
def load_state():
|
|
try:
|
|
with open(CONFIG_FILE, "r") as f:
|
|
return json.load(f)
|
|
except:
|
|
return {"last_ipo_codes": []}
|
|
|
|
def save_state(state):
|
|
with open(CONFIG_FILE, "w") as f:
|
|
json.dump(state, f)
|
|
|
|
def get_ipo_data():
|
|
"""获取港股新股数据"""
|
|
with sync_playwright() as p:
|
|
browser = p.chromium.launch(headless=True)
|
|
page = browser.new_page()
|
|
|
|
page.goto("https://www.jisilu.cn/data/new_stock/#hkipo", timeout=30000)
|
|
page.wait_for_timeout(5000)
|
|
|
|
ipos = []
|
|
|
|
# 查找所有表格
|
|
all_tables = page.query_selector_all("table")
|
|
|
|
for table in all_tables:
|
|
rows = table.query_selector_all("tr")
|
|
|
|
for row in rows[2:]: # 跳过表头
|
|
cols = row.query_selector_all("td")
|
|
if len(cols) >= 15:
|
|
code = cols[0].inner_text().strip()
|
|
name = cols[1].inner_text().strip()
|
|
start_date = cols[3].inner_text().strip()
|
|
end_date = cols[4].inner_text().strip()
|
|
price = cols[9].inner_text().strip() if len(cols) > 9 else "-"
|
|
|
|
# 过滤有效数据
|
|
if code and code.isdigit() and len(code) <= 6:
|
|
ipos.append({
|
|
"code": code,
|
|
"name": name[:15],
|
|
"start_date": start_date,
|
|
"end_date": end_date,
|
|
"price": price
|
|
})
|
|
|
|
browser.close()
|
|
return ipos
|
|
|
|
def check_new(ipos, last_codes):
|
|
current_codes = [ipo["code"] for ipo in ipos]
|
|
new_codes = set(current_codes) - set(last_codes)
|
|
return [ipo for ipo in ipos if ipo["code"] in new_codes]
|
|
|
|
def generate_report(ipos, new_ipos):
|
|
lines = []
|
|
lines.append("="*70)
|
|
lines.append(f"📊 港股新股检查报告")
|
|
lines.append(f"更新时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}")
|
|
lines.append("="*70)
|
|
|
|
if not ipos:
|
|
lines.append("\n⚠️ 暂无港股新股数据")
|
|
else:
|
|
if new_ipos:
|
|
lines.append("\n🎉 【新增申购】")
|
|
lines.append(f"{'代码':<10} {'名称':<16} {'招股价':<12} {'申购起始':<12} {'申购截止':<12}")
|
|
lines.append("-" * 65)
|
|
for ipo in new_ipos:
|
|
lines.append(f"{ipo['code']:<10} {ipo['name']:<16} {ipo['price']:<12} {ipo['start_date']:<12} {ipo['end_date']:<12}")
|
|
else:
|
|
lines.append("\n✅ 【暂无新增申购】")
|
|
|
|
lines.append(f"\n【港股新股 ({len(ipos)}只)】")
|
|
lines.append(f"{'代码':<10} {'名称':<16} {'招股价':<12} {'申购起始':<12} {'申购截止':<12}")
|
|
lines.append("-" * 65)
|
|
for ipo in ipos[:10]:
|
|
lines.append(f"{ipo['code']:<10} {ipo['name']:<16} {ipo['price']:<12} {ipo['start_date']:<12} {ipo['end_date']:<12}")
|
|
|
|
lines.append("\n" + "-"*70)
|
|
lines.append("📍 数据来源: 集思录港股新股")
|
|
lines.append("📍 链接: https://www.jisilu.cn/data/new_stock/#hkipo")
|
|
lines.append("="*70)
|
|
return "\n".join(lines)
|
|
|
|
def main():
|
|
print(f"\n{'='*70}")
|
|
print("📊 港股新股检查")
|
|
print(f"时间: {datetime.now()}")
|
|
print(f"{'='*70}")
|
|
|
|
state = load_state()
|
|
last_codes = state.get("last_ipo_codes", [])
|
|
|
|
print("\n[1] 获取港股新股数据...")
|
|
ipos = get_ipo_data()
|
|
print(f" 获取到 {len(ipos)} 只新股")
|
|
|
|
if ipos:
|
|
print("\n[2] 检查新增...")
|
|
new_ipos = check_new(ipos, last_codes)
|
|
print(f" 新增 {len(new_ipos)} 只")
|
|
|
|
report = generate_report(ipos, new_ipos)
|
|
print(f"\n{report}")
|
|
|
|
current_codes = [ipo["code"] for ipo in ipos]
|
|
save_state({"last_ipo_codes": current_codes})
|
|
else:
|
|
print("\n⚠️ 暂无新股")
|
|
print("请手动查看: https://www.jisilu.cn/data/new_stock/#hkipo")
|
|
|
|
with open(REPORT_FILE, "w") as f:
|
|
f.write(generate_report(ipos, []))
|
|
|
|
print("\n完成!")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|