gao
05635b76b9
状态： - 9 个 sync task（stock_basic / kline_daily / kline_index / kline_5min / moneyflow / industry_sector / sector_features / share_snapshot / market_regime） - 数据源：baostock + mairui + 雪球(pysnowball) + 新浪（4 个） - 项目级约束：永远不用 akshare（已落实） - kline_5min 改用 DB 快照统一全量/增量逻辑 - 零后端 Chrome 扩展 xueqiu_sync（独立项目）
83 lines
2.5 KiB
Python
83 lines
2.5 KiB
Python
"""同步任务管理：手动触发 / 状态查询 / 失败重试。"""
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
|
|
from app.core.sync import get_registry_status
|
|
from app.tasks import get_task, TASKS
|
|
|
|
router = APIRouter(prefix="/api/sync", tags=["sync"])
|
|
|
|
# 防重入：同一 dataset_id 只能有一个 run 在跑
|
|
_running_locks: dict[str, threading.Lock] = {}
|
|
_locks_guard = threading.Lock()
|
|
|
|
|
|
def _get_lock(dataset_id: str) -> threading.Lock:
|
|
with _locks_guard:
|
|
if dataset_id not in _running_locks:
|
|
_running_locks[dataset_id] = threading.Lock()
|
|
return _running_locks[dataset_id]
|
|
|
|
|
|
@router.get("/tasks")
|
|
def list_tasks():
|
|
return {"tasks": list(TASKS.keys())}
|
|
|
|
|
|
@router.get("/registry")
|
|
def registry_status():
|
|
return {"registry": get_registry_status()}
|
|
|
|
|
|
@router.get("/registry/{dataset_id}")
|
|
def registry_one(dataset_id: str):
|
|
rows = get_registry_status()
|
|
for r in rows:
|
|
if r["dataset_id"] == dataset_id:
|
|
return r
|
|
raise HTTPException(404, f"未知 dataset_id: {dataset_id}")
|
|
|
|
|
|
@router.post("/run/{dataset_id}")
|
|
def run_task(dataset_id: str, force: bool = Query(False)):
|
|
"""同步触发一次。如果已在跑且 force=False 则返回 409。"""
|
|
if dataset_id not in TASKS:
|
|
raise HTTPException(404, f"未知 dataset_id: {dataset_id}，可选: {list(TASKS)}")
|
|
lk = _get_lock(dataset_id)
|
|
if not lk.acquire(blocking=False):
|
|
if not force:
|
|
raise HTTPException(409, f"{dataset_id} 正在运行中")
|
|
# force=True 时阻塞等
|
|
lk.acquire()
|
|
|
|
def _do():
|
|
try:
|
|
task = get_task(dataset_id)
|
|
return task.run(trigger_source="manual")
|
|
finally:
|
|
lk.release()
|
|
|
|
thread = threading.Thread(target=_do, daemon=True, name=f"manual-{dataset_id}")
|
|
thread.start()
|
|
return {"status": "accepted", "dataset_id": dataset_id, "trigger": "manual"}
|
|
|
|
|
|
@router.post("/reset/{dataset_id}")
|
|
def reset_status(dataset_id: str):
|
|
"""清除 needs_resync / 错误状态，允许下次调度重试。"""
|
|
if dataset_id not in TASKS:
|
|
raise HTTPException(404, f"未知 dataset_id: {dataset_id}")
|
|
from app.core.db import ops as db_ops
|
|
db_ops.update_dataset_registry_state(
|
|
dataset_id,
|
|
status="idle",
|
|
needs_resync=0,
|
|
last_error=None,
|
|
message="手动重置",
|
|
)
|
|
return {"status": "ok", "dataset_id": dataset_id}
|