#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""verify_akshare_e2e.py — akshare 低频 schtask E2E 验证(VPS 跑)。
验: static/ 各表文件数+最新mtime / events/ per-date 类 parquet / constituent_unified 行数+3source / bs_index_constituent_old 完好。
stdout 行缓冲(ssh 非交互块缓冲兼容)。
"""
import datetime
import sqlite3
import sys
from pathlib import Path

sys.stdout.reconfigure(line_buffering=True)

BASE = Path(r"C:\sanguo_vnpy_v2\data")
ok, warn = [], []

# 1. static/ 各表文件数 + 最新 mtime (akshare_static_download 全部类型都落 data/static/<type>/)
#    per-stock(估值/三表/摘要/北向/股本/十大股东) + per-date(龙虎榜/大宗/两融/解禁) + per-period(预告/快报)
print("===== static/ 各表 (akshare 全类型) =====")
for sub in ["valuation", "financial_abstract", "balance", "income", "cashflow",
            "northbound", "share_capital", "top_holders",
            "dragon_tiger", "block_trade", "margin_sse", "restricted",
            "forecast", "express"]:
    d = BASE / "static" / sub
    if not d.exists():
        warn.append(f"static/{sub} MISSING")
        print(f"  {sub}: MISSING")
        continue
    fs = list(d.glob("*.parquet"))
    if not fs:
        warn.append(f"static/{sub} 0 parquet")
        print(f"  {sub}: 0")
        continue
    mt = max(f.stat().st_mtime for f in fs)
    latest = datetime.datetime.fromtimestamp(mt).strftime("%Y-%m-%d %H:%M")
    print(f"  {sub}: {len(fs)} parquet, latest={latest}")
    ok.append(f"static/{sub}")

# 2. constituent_unified + _old (akshare 全类型已在 section 1 的 static/ 检查覆盖)
print("\n===== constituent_unified + _old =====")
c = sqlite3.connect(str(BASE / "quant_trading.db"), timeout=60)
try:
    c.execute("PRAGMA busy_timeout = 60000")
    n = c.execute("SELECT COUNT(*) FROM constituent_unified").fetchone()[0]
    src = c.execute("SELECT source, COUNT(*) FROM constituent_unified GROUP BY source").fetchall()
    print(f"  constituent_unified: {n} rows, sources={src}")
    if n >= 7110 and len(src) >= 3:
        ok.append("constituent_unified>=7110+3source")
    else:
        warn.append(f"constituent_unified {n}<7110 或 source<3 {src}")
    try:
        nold = c.execute("SELECT COUNT(*) FROM bs_index_constituent_old").fetchone()[0]
        print(f"  bs_index_constituent_old: {nold} rows (baostock 历史时点完好)")
        if nold > 0:
            ok.append("_old intact")
    except Exception as e:
        warn.append(f"_old 查询失败 {e}")
finally:
    c.close()

print("\n===== SUMMARY =====")
print(f"OK: {len(ok)}  WARN: {len(warn)}")
for w in warn:
    print(f"  WARN {w}")
print("VERIFY DONE")