feat(核算)!: 添加 T+ 正式核算与采集托管
补齐源快照、资料预演、正式导入及工资核算的依赖链路。 新增版本化输入、逐段 FIFO、生成锁和结构化当月调整， 并提供默认关闭的周期采集、有限重试及 Windows 托管。 原始客户资料、私有配置及首页改动不纳入本次提交。 BREAKING CHANGE: 升级前需应用新增迁移，并通过 PAYROLL_MAKER_USER_ID 绑定实际工资制作人，否则业务只读。 旧单批次核算默认关闭；历史兼容需显式启用 PAYROLL_ALLOW_LEGACY_CALCULATION。生产启用和真实金额待验收。
This commit is contained in:
liuwenxin committed
1 parent
1b074576fe
commit
108bf87225
154 files changed
+9515
-645
No files matched your search
@@ -0,0 +1,273 @@
|
||||
"""周期检查只排队源采集：稳定意图键、有限重试、独立心跳和只读异常概况。"""
|
||||
|
||||
import hashlib
|
||||
import logging
|
||||
import threading
|
||||
from contextlib import contextmanager
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from uuid import NAMESPACE_URL, uuid4, uuid5
|
||||
|
||||
from sqlalchemy import func, inspect, select, text
|
||||
from sqlalchemy.dialects.mysql import insert
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from app.core.config import get_settings
|
||||
from app.core.tplus_db import resolve_tplus_settings
|
||||
from app.models import TPlusAutomationState, TPlusSyncBatch, TPlusSyncError, TPlusWorkRun
|
||||
from app.services.tplus import task_service
|
||||
from app.services.tplus.errors import TRANSIENT_CODES
|
||||
from app.services.tplus.mapping import DETAIL_MAPPINGS, TPLUS_MAPPINGS
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
SHANGHAI = timezone(timedelta(hours=8))
|
||||
ENTITIES = tuple(sorted(m.entity_type for m in (*TPLUS_MAPPINGS, *DETAIL_MAPPINGS)))
|
||||
|
||||
|
||||
def local_now():
|
||||
return datetime.now(SHANGHAI).replace(tzinfo=None)
|
||||
|
||||
|
||||
def source_identity(settings):
|
||||
source = resolve_tplus_settings(settings)
|
||||
value = f"{source.t_plus_host.casefold()}:{source.t_plus_port}/{source.t_plus_database}"
|
||||
return hashlib.sha256(value.encode()).hexdigest(), source.t_plus_database
|
||||
|
||||
|
||||
def due_period(now, weekday=0, clock="02:00"):
|
||||
"""取最近一次已到期时段；只补这一个周期，不逐周补发历史任务。"""
|
||||
if now.tzinfo:
|
||||
now = now.astimezone(SHANGHAI).replace(tzinfo=None)
|
||||
hour, minute = map(int, clock.split(":"))
|
||||
day = now.date() - timedelta(days=(now.weekday() - weekday) % 7)
|
||||
due = datetime.combine(day, datetime.min.time()).replace(hour=hour, minute=minute)
|
||||
if due > now:
|
||||
due -= timedelta(days=7)
|
||||
year, week, _ = due.isocalendar()
|
||||
return f"{year}-W{week:02d}", due
|
||||
|
||||
|
||||
def intent_key(source_key, period, attempt=0):
|
||||
return str(uuid5(NAMESPACE_URL, f"payroll:tplus:weekly:{source_key}:{period}:{attempt}"))
|
||||
|
||||
|
||||
def _state(session, key, database, *, lock=False):
|
||||
if lock and session.bind.dialect.name == "mysql":
|
||||
stmt = insert(TPlusAutomationState).values(source_key=key, source_database=database)
|
||||
session.execute(stmt.on_duplicate_key_update(source_key=stmt.inserted.source_key))
|
||||
statement = select(TPlusAutomationState).where(TPlusAutomationState.source_key == key)
|
||||
if lock:
|
||||
statement = statement.with_for_update().execution_options(populate_existing=True)
|
||||
row = session.scalar(statement)
|
||||
if row is None and lock:
|
||||
row = TPlusAutomationState(source_key=key, source_database=database, retry_attempts=0, worker_phase="STOPPED")
|
||||
session.add(row)
|
||||
session.flush()
|
||||
return row
|
||||
|
||||
|
||||
@contextmanager
|
||||
def maintenance_lock(session, key):
|
||||
"""维护检查独立于长采集锁；同一源的两个检查不能重复创建重试。"""
|
||||
if session.bind.dialect.name != "mysql":
|
||||
yield True
|
||||
return
|
||||
connection = session.get_bind().connect()
|
||||
name = "ys:tplus:maintain:" + key[:32]
|
||||
acquired = False
|
||||
try:
|
||||
acquired = connection.execute(text("SELECT GET_LOCK(:name,0)"), {"name": name}).scalar() == 1
|
||||
yield acquired
|
||||
finally:
|
||||
if acquired:
|
||||
try:
|
||||
if connection.execute(text("SELECT RELEASE_LOCK(:name)"), {"name": name}).scalar() != 1:
|
||||
connection.invalidate()
|
||||
except Exception:
|
||||
connection.invalidate()
|
||||
connection.close()
|
||||
|
||||
|
||||
def _has_active(session):
|
||||
# 实际读取仍受原worker命名锁保护；排队检查防止周期任务积压。
|
||||
for model in (TPlusSyncBatch, TPlusWorkRun):
|
||||
if session.scalar(select(model.id).where(model.status.in_(("QUEUED", "RUNNING"))).limit(1)) is not None:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _enqueue(session, state, period, attempt, parent=None):
|
||||
values = {
|
||||
"batch_no": task_service.new_batch_no(), "request_key": intent_key(state.source_key, period, attempt),
|
||||
"sync_mode": "FULL", "status": "QUEUED", "source_database": state.source_database,
|
||||
"entity_types_json": list(ENTITIES), "parent_batch_no": parent,
|
||||
"trigger_type": "AUTO_RETRY" if attempt else "SCHEDULED", "automatic_attempt": attempt,
|
||||
"source_record_count": 0, "staged_record_count": 0, "failed_record_count": 0, "skipped_record_count": 0,
|
||||
}
|
||||
batch, inserted = task_service._insert(session, values)
|
||||
if inserted and parent:
|
||||
from app.repositories.tplus_repository import increment_retry_errors
|
||||
|
||||
increment_retry_errors(session, parent)
|
||||
state.scheduled_period = period
|
||||
state.current_batch_no = batch.batch_no
|
||||
state.retry_attempts = attempt
|
||||
state.next_retry_at = None
|
||||
if not attempt:
|
||||
state.root_batch_no = batch.batch_no
|
||||
return batch
|
||||
|
||||
|
||||
def _cycle_current(session, state):
|
||||
"""锁住本周期重试树；人工恢复也算本周期进度，自动次数不能被重置。"""
|
||||
if not state.root_batch_no:
|
||||
return None
|
||||
root = session.scalar(select(TPlusSyncBatch).where(
|
||||
TPlusSyncBatch.batch_no == state.root_batch_no,
|
||||
).with_for_update().execution_options(populate_existing=True))
|
||||
if root is None:
|
||||
return None
|
||||
current, frontier, seen = root, [root.batch_no], {root.batch_no}
|
||||
while frontier:
|
||||
children = list(session.scalars(select(TPlusSyncBatch).where(
|
||||
TPlusSyncBatch.parent_batch_no.in_(frontier),
|
||||
TPlusSyncBatch.source_database == state.source_database,
|
||||
).order_by(TPlusSyncBatch.id).with_for_update().execution_options(populate_existing=True)))
|
||||
frontier = []
|
||||
for child in children:
|
||||
if child.batch_no in seen:
|
||||
continue
|
||||
seen.add(child.batch_no)
|
||||
frontier.append(child.batch_no)
|
||||
state.retry_attempts = max(state.retry_attempts or 0, child.automatic_attempt or 0)
|
||||
if child.id > current.id:
|
||||
current = child
|
||||
state.current_batch_no = current.batch_no
|
||||
return current
|
||||
|
||||
|
||||
def tick(session, *, settings=None, now=None):
|
||||
settings, now = settings or get_settings(), now or local_now()
|
||||
if not settings.t_plus_automation_enabled:
|
||||
return {"status": "DISABLED", "queued": False}
|
||||
if settings.t_plus_stop_file and Path(settings.t_plus_stop_file).exists():
|
||||
return {"status": "STOP_REQUESTED", "queued": False}
|
||||
key, database = source_identity(settings)
|
||||
with maintenance_lock(session, key) as acquired:
|
||||
if not acquired:
|
||||
return {"status": "BUSY", "queued": False}
|
||||
session.rollback()
|
||||
state = _state(session, key, database, lock=True)
|
||||
state.maintenance_at = now
|
||||
period, _ = due_period(now, settings.t_plus_schedule_weekday, settings.t_plus_schedule_time)
|
||||
current = _cycle_current(session, state)
|
||||
if current and current.status == "SUCCEEDED":
|
||||
state.last_success_at = current.finished_at
|
||||
state.next_retry_at = None
|
||||
if _has_active(session):
|
||||
session.commit()
|
||||
return {"status": "DEFERRED", "queued": False, "period": period}
|
||||
batch = None
|
||||
if state.scheduled_period != period:
|
||||
batch = _enqueue(session, state, period, 0)
|
||||
elif current and current.status == "FAILED" and (state.retry_attempts or 0) < 2:
|
||||
codes = set(session.scalars(select(TPlusSyncError.error_code).where(TPlusSyncError.batch_no == current.batch_no)))
|
||||
if codes and codes <= TRANSIENT_CODES and current.finished_at is not None:
|
||||
delay = (5, 15)[state.retry_attempts or 0]
|
||||
state.next_retry_at = current.finished_at + timedelta(minutes=delay)
|
||||
if now >= state.next_retry_at:
|
||||
batch = _enqueue(session, state, period, (state.retry_attempts or 0) + 1, current.batch_no)
|
||||
else:
|
||||
state.next_retry_at = None
|
||||
elif current and current.status in ("FAILED", "PARTIAL"):
|
||||
state.next_retry_at = None
|
||||
session.commit()
|
||||
return {"status": "QUEUED" if batch else "CHECKED", "queued": batch is not None,
|
||||
"batch_no": batch.batch_no if batch else None, "period": period}
|
||||
|
||||
|
||||
def heartbeat(session_factory, settings, owner, phase, *, now=None, claim=False):
|
||||
key, database = source_identity(settings)
|
||||
with session_factory() as session:
|
||||
state = _state(session, key, database, lock=True)
|
||||
if state.worker_owner not in (None, owner) and not claim:
|
||||
session.rollback()
|
||||
return
|
||||
state.worker_owner, state.worker_phase = owner, phase
|
||||
state.heartbeat_at = now or local_now()
|
||||
session.commit()
|
||||
|
||||
|
||||
class WorkerHeartbeat:
|
||||
"""独立会话和线程写心跳，长查询不阻塞；只在worker已持采集锁时启动。"""
|
||||
def __init__(self, session, phase="RUNNING", interval=30, owner=None):
|
||||
self.factory = sessionmaker(session.get_bind(), expire_on_commit=False)
|
||||
self.settings = get_settings()
|
||||
self.owner, self.phase, self.interval = owner or str(uuid4()), phase, interval
|
||||
self.stop = threading.Event()
|
||||
self.thread = None
|
||||
|
||||
def __enter__(self):
|
||||
try:
|
||||
source_identity(self.settings)
|
||||
with self.factory() as session:
|
||||
if not inspect(session.connection()).has_table(TPlusAutomationState.__tablename__):
|
||||
return self
|
||||
heartbeat(self.factory, self.settings, self.owner, self.phase, claim=True)
|
||||
except Exception:
|
||||
logger.warning("T+心跳暂不可写，请检查迁移与运行配置")
|
||||
return self
|
||||
self.thread = threading.Thread(target=self._pulse, name="tplus-heartbeat", daemon=True)
|
||||
self.thread.start()
|
||||
return self
|
||||
|
||||
def _pulse(self):
|
||||
while not self.stop.wait(self.interval):
|
||||
try:
|
||||
heartbeat(self.factory, self.settings, self.owner, self.phase)
|
||||
except Exception:
|
||||
logger.warning("T+进程心跳更新失败")
|
||||
|
||||
def __exit__(self, *_args):
|
||||
self.stop.set()
|
||||
if self.thread:
|
||||
self.thread.join(timeout=5)
|
||||
try:
|
||||
heartbeat(self.factory, self.settings, self.owner, "IDLE")
|
||||
except Exception:
|
||||
logger.warning("T+进程状态更新失败")
|
||||
|
||||
|
||||
def overview(session, *, settings=None, now=None):
|
||||
settings, now = settings or get_settings(), now or local_now()
|
||||
base = {"enabled": settings.t_plus_automation_enabled, "weekday": settings.t_plus_schedule_weekday,
|
||||
"time": settings.t_plus_schedule_time, "timezone": "Asia/Shanghai", "worker_phase": "UNKNOWN",
|
||||
"heartbeat_at": None, "last_success_at": None, "next_retry_at": None, "alerts": []}
|
||||
base["stop_requested"] = bool(settings.t_plus_stop_file and Path(settings.t_plus_stop_file).exists())
|
||||
try:
|
||||
key, database = source_identity(settings)
|
||||
except RuntimeError:
|
||||
return base
|
||||
if not inspect(session.connection()).has_table(TPlusAutomationState.__tablename__):
|
||||
base["alerts"] = [{"code": "MIGRATION_REQUIRED", "message": "运行状态结构尚未升级"}]
|
||||
return base
|
||||
state = _state(session, key, database)
|
||||
if state:
|
||||
base.update(worker_phase=state.worker_phase, heartbeat_at=state.heartbeat_at,
|
||||
last_success_at=state.last_success_at, next_retry_at=state.next_retry_at)
|
||||
alerts = base["alerts"]
|
||||
latest = session.scalar(select(TPlusSyncBatch).where(TPlusSyncBatch.source_database == database).order_by(TPlusSyncBatch.id.desc()).limit(1))
|
||||
if latest and latest.status in ("FAILED", "PARTIAL"):
|
||||
alerts.append({"code": "SYNC_REQUIRES_ATTENTION", "message": "最近采集失败或部分失败，请核对后处理", "batch_no": latest.batch_no})
|
||||
queued = session.scalar(select(TPlusSyncBatch).where(TPlusSyncBatch.source_database == database,
|
||||
TPlusSyncBatch.status == "QUEUED", TPlusSyncBatch.create_time <= now - timedelta(minutes=15)).order_by(TPlusSyncBatch.id).limit(1))
|
||||
if queued:
|
||||
alerts.append({"code": "QUEUE_OVERDUE", "message": "采集任务排队超过15分钟", "batch_no": queued.batch_no})
|
||||
last_success = session.scalar(select(func.max(TPlusSyncBatch.finished_at)).where(TPlusSyncBatch.source_database == database, TPlusSyncBatch.status == "SUCCEEDED", TPlusSyncBatch.sync_mode == "FULL"))
|
||||
base["last_success_at"] = last_success
|
||||
if settings.t_plus_automation_enabled and not base["stop_requested"]:
|
||||
if last_success is None or now - last_success > timedelta(days=8):
|
||||
alerts.append({"code": "SYNC_STALE", "message": "尚无成功采集或超过8天未成功同步"})
|
||||
if state is None or state.heartbeat_at is None or now - state.heartbeat_at > timedelta(minutes=2) or state.worker_phase == "STOPPED":
|
||||
alerts.append({"code": "WORKER_OFFLINE", "message": "采集进程心跳中断，请检查托管任务"})
|
||||
return base
|
||||
Reference in new issue
Block a user