130 lines
6.4 KiB
Python
130 lines
6.4 KiB
Python
"""按租户和派车日期补派历史记录，并刷新已有自动现场记录。
|
|
|
|
这是运维脚本，使用显式租户范围和专用审计身份；不提供 HTTP 入口。
|
|
默认只读预览，--apply 才写入。历史记录仅在命中现有派发规则时启用后续同步。
|
|
在 backend 目录执行，例如：
|
|
venv/bin/python scripts/refresh_dispatch_calendar_site.py --env-file .env \
|
|
--tenant shanxia2.0 --start-date 2026-10-28 --end-date 2026-10-30
|
|
增加 --apply 正式执行；生产配置使用 --env-file .env.production。
|
|
"""
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from dotenv import dotenv_values
|
|
from sqlalchemy import create_engine, or_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.db_auth import DbUserContext
|
|
from app.models.pl_dispatch_calendar import PlDispatchCalendar as Record
|
|
from app.services import pl_dispatch_calendar_service as svc
|
|
|
|
SOURCE_KINDS = ('acid', 'alkali', 'other')
|
|
COPY_FIELDS = ('dispatch_date', 'vehicle_id', 'license_plate', 'product_id', 'product_name',
|
|
'origin', 'destination', 'quantity', 'remark')
|
|
|
|
|
|
def refresh(db: Session, *, tenant: str, start: date, end: date,
|
|
operator: str, apply: bool = False) -> dict:
|
|
"""一次事务内补派；不改手工现场行，不按日期/车牌猜测关联。"""
|
|
if not tenant.strip() or not operator.strip() or len(operator) > 64 or start > end:
|
|
raise ValueError('须提供租户、有效审计身份及正确日期范围')
|
|
if not date(1900, 1, 1) <= start <= end <= date(2100, 12, 31):
|
|
raise ValueError('派车日期须在 1900-01-01 至 2100-12-31 范围内')
|
|
user = DbUserContext(operator, '现场日历补派脚本', '', 'super_admin', tenant)
|
|
result = {'tenant': tenant, 'startDate': start.isoformat(), 'endDate': end.isoformat(),
|
|
'apply': apply, 'counts': {key: 0 for key in
|
|
('created', 'updated', 'restored', 'removed', 'unchanged', 'ignored', 'enabled')},
|
|
'records': []}
|
|
try:
|
|
query = db.query(Record).filter(
|
|
Record.tenant_no == tenant, Record.calendar_kind.in_(SOURCE_KINDS),
|
|
Record.dispatch_date >= start, Record.dispatch_date <= end,
|
|
or_(Record.del_flag == 0, Record.site_sync_enabled == 1),
|
|
).order_by(Record.id)
|
|
if apply:
|
|
query = query.populate_existing().with_for_update()
|
|
sources = query.all()
|
|
for source in sources:
|
|
target_query = db.query(Record).filter(
|
|
Record.tenant_no == tenant, Record.source_record_id == source.id,
|
|
)
|
|
if apply:
|
|
target_query = target_query.populate_existing().with_for_update()
|
|
target = target_query.one_or_none()
|
|
if target is not None and target.calendar_kind != 'site':
|
|
raise ValueError('来源关联到了非现场日历，停止刷新：' + source.id)
|
|
document_type = None if source.del_flag == 1 else svc._site_document_type(
|
|
source.origin, source.destination)
|
|
enable = not source.site_sync_enabled and document_type is not None
|
|
if not document_type:
|
|
action = 'removed' if target is not None and target.del_flag == 0 else 'ignored'
|
|
elif target is None:
|
|
action = 'created'
|
|
elif target.del_flag == 1:
|
|
action = 'restored'
|
|
elif target.document_type != document_type or any(
|
|
getattr(target, field) != getattr(source, field) for field in COPY_FIELDS):
|
|
action = 'updated'
|
|
else:
|
|
action = 'unchanged'
|
|
changes_target = action in ('created', 'updated', 'restored', 'removed')
|
|
# 预览与正式执行一致：异常关联须人工核对，不能静默越过同步开关。
|
|
if changes_target and not (source.site_sync_enabled or enable):
|
|
raise ValueError('自动关联存在但来源未启用同步，停止刷新：' + source.id)
|
|
result['counts'][action] += 1
|
|
result['counts']['enabled'] += int(enable)
|
|
if apply:
|
|
if enable:
|
|
source.site_sync_enabled = 1
|
|
source.update_by = operator
|
|
if changes_target:
|
|
svc._sync_site_record(db, tenant, user, source)
|
|
db.flush()
|
|
result['records'].append({'sourceId': source.id, 'action': action,
|
|
'enabled': bool(enable)})
|
|
if apply:
|
|
db.commit()
|
|
else:
|
|
db.rollback()
|
|
return result
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
parser.add_argument('--env-file', required=True, help='明确指定 .env 或 .env.production')
|
|
parser.add_argument('--tenant', required=True)
|
|
parser.add_argument('--start-date', type=date.fromisoformat, required=True)
|
|
parser.add_argument('--end-date', type=date.fromisoformat, required=True)
|
|
parser.add_argument('--operator', default='calendar-site-refresh', help='写入 update_by 的运维审计身份')
|
|
parser.add_argument('--apply', action='store_true', help='提交变更，默认只读预览')
|
|
args = parser.parse_args()
|
|
# 直接读取指定文件，避免进程环境 DATABASE_URL 覆盖所选数据库。
|
|
url = dotenv_values(args.env_file).get('DATABASE_URL')
|
|
if not url:
|
|
parser.error('指定配置文件没有 DATABASE_URL')
|
|
engine = create_engine(url, pool_pre_ping=True, connect_args={
|
|
'connect_timeout': 10, 'read_timeout': 60, 'write_timeout': 60})
|
|
try:
|
|
with Session(engine, autoflush=False, expire_on_commit=False) as db:
|
|
if engine.dialect.name == 'mysql':
|
|
from sqlalchemy import text
|
|
db.execute(text('SET SESSION innodb_lock_wait_timeout = 15'))
|
|
result = refresh(db, tenant=args.tenant, start=args.start_date, end=args.end_date,
|
|
operator=args.operator, apply=args.apply)
|
|
print(json.dumps({'database': engine.url.database, 'host': engine.url.host, **result},
|
|
ensure_ascii=False, indent=2))
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|