auto_control/scripts/tests/test_upload_name.py

# -*- coding: utf-8 -*-
"""P1 集成测试：按名字上传/配标题走通（**临时 SQLite + 临时视频目录**，不碰真实数据）。"""
import io
import os
import shutil
import sys
import tempfile

sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')

TMP = tempfile.mkdtemp(prefix='ac_p1_')
os.environ['DATABASE_URL'] = 'sqlite:///' + os.path.join(TMP, 'ac.db').replace('\\', '/')
os.environ['DATA_VIDEO_DIR'] = os.path.join(TMP, 'videos')
os.environ['DEPLOY_ENV'] = 'dev'
os.environ['DB_ALLOW_SQLITE_FALLBACK'] = '1'

# 仓库根 = scripts/tests/ 往上三级（不写死绝对路径，换机器也能跑）
_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, _ROOT)
os.chdir(_ROOT)

from flask import Flask                                              # noqa: E402
from sqlalchemy import inspect                                       # noqa: E402
from core import db_config, models                                   # noqa: E402
from core import video_plan as vp                                    # noqa: E402

fails = []


def check(label, got, want):
    ok = got == want
    if not ok:
        fails.append(label)
    print(('  ✅ ' if ok else '  ❌ ') + label + ('' if ok else '\n       得到 %r\n       期望 %r' % (got, want)))


def ok(label, cond, detail=''):
    if not cond:
        fails.append(label)
    print(('  ✅ ' if cond else '  ❌ ') + label + ('' if cond else '  ' + str(detail)))


class FakeFile:
    """够用的 FileStorage 替身：只用到 .filename / .stream。"""

    def __init__(self, filename, data):
        self.filename = filename
        self.stream = io.BytesIO(data)


app = Flask("t")
uri = db_config.build_db_uri()
print('测试库:', uri)
app.config["SQLALCHEMY_DATABASE_URI"] = uri
app.config["SQLALCHEMY_ENGINE_OPTIONS"] = db_config.engine_options(uri)
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
models.db.init_app(app)
with app.app_context():
    models.db.create_all()
    models._ensure_plain_indexes()        # 生产启动路径里的那一步（老库补索引）
vp.init_app(app)

with app.app_context():
    print('=== 0. 前提：台账一张空表也不依赖 ===')
    check('device_account 里 0 行（上传完全不查台账）', models.DeviceAccount.query.count(), 0)
    check('video_plan 里 0 行', models.VideoPlan.query.count(), 0)

    print('=== 1. 上传：名字_日期_编号 ===')
    r = vp.upload_video(FakeFile('小王_20260912_1.mp4', b'A' * 500))
    ok('上传成功', r.get('ok') is True, r)
    p = r['plan']
    check('昵称 = 文件名里的名字', p['nickname'], '小王')
    check('手机号留空（不再写台账快照）', p['phone'], '')
    check('设备号留空', p['device_name'], '')
    check('发布日期', p['release_date'], '2026-09-12')
    check('编号', p['seq'], 1)
    check('编号不是自动的', p['seq_auto'], False)
    check('状态 = 待标题', p['status'], 'pending')

    print('=== 2. 幂等：同一份内容重传 ===')
    r2 = vp.upload_video(FakeFile('小王_20260912_1.mp4', b'A' * 500))
    check('跳过', r2.get('action'), 'skip')

    print('=== 3. 同槽位换内容：先拒收、replace 才行 ===')
    r3 = vp.upload_video(FakeFile('小王_20260912_1.mp4', b'B' * 600))
    ok('拒收', r3.get('ok') is False and '已有视频' in (r3.get('error') or ''), r3)
    r4 = vp.upload_video(FakeFile('小王_20260912_1.mp4', b'B' * 600), replace=True)
    check('覆盖成功', r4.get('action'), 'replace')
    check('还是 1 行', models.VideoPlan.query.count(), 1)

    print('=== 4. 不带编号 → 自动排到空位 ===')
    r5 = vp.upload_video(FakeFile('小王_20260912.mp4', b'C' * 700))
    check('自动编号 = 2', r5['plan']['seq'], 2)
    check('标记为自动', r5['plan']['seq_auto'], True)

    print('=== 5. 老格式（手机号）被拒 + 说清怎么改 ===')
    r6 = vp.upload_video(FakeFile('13382508423_20261011_01.mp4', b'D' * 700))
    ok('拒收且提示是手机号', r6.get('ok') is False and '手机号' in (r6.get('error') or ''), r6)
    check('库里还是 2 行', models.VideoPlan.query.count(), 2)

    print('=== 6. 配标题（`标题 || 正文_名字_日期_编号`）===')
    t1 = vp.import_titles('聚划算 || 今天全场五折，点进来看看_小王_20260912')
    check('配到 1 条', t1['counts']['attached'], 1)
    row1 = models.VideoPlan.query.filter_by(nickname='小王', release_date='2026-09-12',
                                            seq=1).first()
    check('标题落库', row1.title, '聚划算')
    check('正文落库', row1.body, '今天全场五折，点进来看看')
    check('状态自动切到待发布', row1.status, 'ready')
    check('返回体里键是 name', t1['rows'][0]['name'], '小王')

    t2 = vp.import_titles('第二个 || 正文二_小王_20260912_2')
    check('按编号精确配对', t2['counts']['attached'], 1)
    row2 = models.VideoPlan.query.filter_by(nickname='小王', release_date='2026-09-12',
                                            seq=2).first()
    check('第 2 条标题', row2.title, '第二个')

    t3 = vp.import_titles('这条没有主_不存在的名字_20260912')
    check('认不出名字 → 拒收', (t3['counts']['rejected'], t3['rows'][0]['error']),
          (1, '该名字 2026-09-12 没有待配标题的视频（请先上传视频）'))

    # 带编号重配同一条 → 认出来是"配过了"，跳过（幂等）
    t4 = vp.import_titles('聚划算 || 今天全场五折，点进来看看_小王_20260912_1')
    check('带编号重配同一条 → 跳过', t4['counts']['skipped'], 1)
    # 不带编号：它只会找"还没有标题的行"，两行都配过 → 拒收（既有语义，文案说清了原因）
    t5 = vp.import_titles('聚划算 || 今天全场五折，点进来看看_小王_20260912')
    check('不带编号且两行都有标题 → 拒收', (t5['counts']['rejected'], t5['rows'][0]['error']),
          (1, '该名字 2026-09-12 没有待配标题的视频（请先上传视频）'))

    print('=== 7. 老命名空间不受影响：按名字查重/取槽位 ===')
    check('taken_slots 按名字', vp.taken_slots('小王', '2026-09-12'), {1, 2})
    check('别的名字是空的', vp.taken_slots('小李', '2026-09-12'), set())

    print('=== 8. 索引补建 ===')
    idx = {i['name'] for i in inspect(models.db.engine).get_indexes('video_plan')}
    ok('ix_video_plan_name_date 已建', 'ix_video_plan_name_date' in idx, sorted(idx))

    print('=== 9. 素材落盘 ===')
    files = os.listdir(os.path.join(TMP, 'videos', '2026-10'))
    check('落盘 2 个文件（覆盖后旧文件已删）', len(files), 2)

print()
shutil.rmtree(TMP, ignore_errors=True)
if fails:
    print('❌ 失败 %d 项：%s' % (len(fails), ' / '.join(fails)))
    sys.exit(1)
print('✅ 全部通过（临时库已清理）')