95 lines
3.2 KiB
Python
95 lines
3.2 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
sys.path.insert(0, str(ROOT / "backend"))
|
||
|
|
|
||
|
|
from app.main import ( # noqa: E402
|
||
|
|
accounts,
|
||
|
|
generate_daily_report,
|
||
|
|
get_cls_news_payload,
|
||
|
|
get_daily_inputs,
|
||
|
|
get_opinion_report,
|
||
|
|
get_report_list,
|
||
|
|
health,
|
||
|
|
put_daily_inputs,
|
||
|
|
)
|
||
|
|
from app.models import DailyInputUpsertPayload # noqa: E402
|
||
|
|
from app.services.storage import ( # noqa: E402
|
||
|
|
CLS_NEWS_PATH,
|
||
|
|
daily_input_path,
|
||
|
|
ensure_data_dirs,
|
||
|
|
report_path,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def assert_true(condition: bool, message: str) -> None:
|
||
|
|
if not condition:
|
||
|
|
raise AssertionError(message)
|
||
|
|
|
||
|
|
|
||
|
|
def run() -> None:
|
||
|
|
ensure_data_dirs()
|
||
|
|
|
||
|
|
date_str = "2026-03-19"
|
||
|
|
payload = DailyInputUpsertPayload(
|
||
|
|
accounts=[
|
||
|
|
{
|
||
|
|
"account_id": "touzi-mingjian",
|
||
|
|
"links": ["https://mp.weixin.qq.com/s/_l429HDdGFi18eOJpDujjA"],
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"account_id": "aigujun-2020",
|
||
|
|
"links": ["https://mp.weixin.qq.com/s/1No9toallxkKRjpj4wZt9Q"],
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"account_id": "mazhiming-shouping",
|
||
|
|
"links": ["https://mp.weixin.qq.com/s/i0vlr02f7Ydb9GvCa7idTw"],
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"account_id": "laobai-guandian",
|
||
|
|
"links": ["https://mp.weixin.qq.com/s/hRqoxqoR8UNiWw3UEj6_yQ"],
|
||
|
|
},
|
||
|
|
]
|
||
|
|
)
|
||
|
|
|
||
|
|
saved_input = put_daily_inputs(date_str, payload)
|
||
|
|
generated_report = generate_daily_report(date_str)
|
||
|
|
loaded_input = get_daily_inputs(date_str)
|
||
|
|
loaded_report = get_opinion_report(date_str)
|
||
|
|
report_list = get_report_list()
|
||
|
|
cls_news = get_cls_news_payload()
|
||
|
|
|
||
|
|
assert_true(health()["status"] == "ok", "health endpoint failed")
|
||
|
|
assert_true(len(accounts()) == 4, "account count should be 4")
|
||
|
|
assert_true(len(saved_input.accounts) == 4, "saved daily input should contain 4 accounts")
|
||
|
|
assert_true(len(loaded_input.accounts) == 4, "loaded daily input should contain 4 accounts")
|
||
|
|
assert_true(generated_report.article_count == 4, "generated report should contain 4 articles")
|
||
|
|
assert_true(loaded_report.article_count == 4, "loaded report should contain 4 articles")
|
||
|
|
assert_true(any(item.date == date_str for item in report_list), "report list should include target date")
|
||
|
|
assert_true(len(cls_news.items) > 0, "cls news should not be empty")
|
||
|
|
assert_true(daily_input_path(date_str).exists(), "daily input json file missing")
|
||
|
|
assert_true(report_path(date_str).exists(), "report json file missing")
|
||
|
|
assert_true(CLS_NEWS_PATH.exists(), "cls news json file missing")
|
||
|
|
|
||
|
|
summary = {
|
||
|
|
"health": health(),
|
||
|
|
"accounts": [item.name for item in accounts()],
|
||
|
|
"daily_input_file": str(daily_input_path(date_str)),
|
||
|
|
"report_file": str(report_path(date_str)),
|
||
|
|
"cls_news_file": str(CLS_NEWS_PATH),
|
||
|
|
"daily_input_accounts": len(loaded_input.accounts),
|
||
|
|
"report_article_count": loaded_report.article_count,
|
||
|
|
"report_focus_sectors": loaded_report.focus_sectors,
|
||
|
|
"report_dates": [item.date for item in report_list[:5]],
|
||
|
|
"cls_news_items": len(cls_news.items),
|
||
|
|
}
|
||
|
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
run()
|