Jayway Liu
509b406254
- F01: _safe_unlink_export 不跟随符号链接、逐层校验父目录，导出失败路径与 清理命令统一使用，只清理受控目录内本次创建的文件 - F02: 清理失败记录告警日志；清理命令回收 failed 任务残留文件 - F05: 定义保存事务内锁数据集行，配额原子检查；revision 冲突返回 409； 保留最近 20 版历史；支持个人/团队共享；审计与变更同事务 - F06: 工作簿参数页毛利率改为只读快照；可编辑单元格解锁并加数值校验 - F07: 对账按指标分容差（人数 0、比率 0.0001、金额 0.01）；自定义列使用 服务端 Decimal 表达式求值参考 - F08: 导出任务执行/下载阶段重校验账号有效性与角色 - 对话草稿会话恢复；载入定义后追问携带原定义（F04）；下载改浏览器直传 - 迁移 0006-0008；PostgreSQL 咨询锁语义修正（import_worker）
75 lines
2.8 KiB
Python
75 lines
2.8 KiB
Python
import time
|
|
|
|
from django.core.management.base import BaseCommand
|
|
from django.db import connection, transaction
|
|
from django.utils import timezone
|
|
|
|
from finance.exporting import process_formula_export
|
|
from finance.importing import process_dataset
|
|
from finance.models import Dataset, FormulaExport
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = "Process durable import and formula-export jobs"
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument("--once", action="store_true")
|
|
|
|
def handle(self, *args, **options):
|
|
postgres = connection.vendor == "postgresql"
|
|
if postgres:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("SELECT pg_try_advisory_lock(782146291)")
|
|
if not cursor.fetchone()[0]:
|
|
self.stdout.write("Another worker owns the database lock")
|
|
return
|
|
try:
|
|
self.run_jobs(options, True)
|
|
finally:
|
|
if postgres:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("SELECT pg_advisory_unlock(782146291)")
|
|
|
|
def run_jobs(self, options, recover):
|
|
while True:
|
|
processed = self._next_import(recover)
|
|
if not processed:
|
|
processed = self._next_export(recover)
|
|
if options["once"]:
|
|
return
|
|
if not processed:
|
|
time.sleep(2)
|
|
|
|
def _claim(self, model, states):
|
|
with transaction.atomic():
|
|
qs = model.objects.filter(status__in=states).order_by("created_at")
|
|
if connection.features.has_select_for_update_skip_locked:
|
|
qs = qs.select_for_update(skip_locked=True)
|
|
item = qs.first()
|
|
if not item:
|
|
return None
|
|
updated = model.objects.filter(pk=item.pk, status__in=states).update(status="processing")
|
|
return item if updated else None
|
|
|
|
def _next_import(self, recover):
|
|
states = ["queued", "processing"] if recover else ["queued"]
|
|
item = self._claim(Dataset, states)
|
|
if not item:
|
|
return False
|
|
Dataset.objects.filter(pk=item.pk).update(started_at=timezone.now())
|
|
result = process_dataset(item.pk)
|
|
self.stdout.write(f"{result.pk}: {result.status}, {result.row_count} rows")
|
|
return True
|
|
|
|
def _next_export(self, recover):
|
|
states = ["queued", "processing"] if recover else ["queued"]
|
|
item = self._claim(FormulaExport, states)
|
|
if not item:
|
|
return False
|
|
try:
|
|
stats = process_formula_export(item.pk)
|
|
self.stdout.write(f"{item.pk}: formula export ready, {stats['rows']} rows")
|
|
except Exception as exc:
|
|
self.stderr.write(f"Formula export {item.pk} failed: {exc}")
|
|
return True
|