# -*- coding: utf-8 -*- """P2 测试：读名字 / 按名字取计划 / 推送前核对（**临时库 + 假设备**，不碰真实数据、不连真机）。""" 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_p2_') 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 core import db_config, models # noqa: E402 from core import video_plan as vp # noqa: E402 from tasks.generic import task as gt # noqa: E402 from tasks.generic import publish_flow as flow # 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 StubEl: """假元素：u2 的 XPathSelector / UiObject 各用到的成员都给上。""" def __init__(self, text): self._t = text @property def exists(self): return self._t is not None def wait(self, timeout=0): return self._t is not None def get_text(self, timeout=None): return self._t class StubD: def __init__(self, text): self.text = text def xpath(self, v): return StubEl(self.text) app = Flask("t") uri = db_config.build_db_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() vp.init_app(app) print('=== 1. clean_account_name（纯函数）===') check('只有名字', vp.clean_account_name('小王'), '小王') check('多行：取第一行', vp.clean_account_name('小王\n编辑资料\n抖音号：35377983067'), '小王') check('首行是标签 → 往下找', vp.clean_account_name('编辑资料\n小王'), '小王') check('抖音号行跳过', vp.clean_account_name('抖音号：35377983067\n小王'), '小王') check('统计行跳过', vp.clean_account_name('关注\n粉丝\n小王'), '小王') check('名字里的双空格保留', vp.clean_account_name('有牛奶面包 你吃吗？'), '有牛奶面包 你吃吗？') check('名字带表情保留', vp.clean_account_name('🐟小鱼'), '🐟小鱼') check('「关注我领福利」不当标签', vp.clean_account_name('关注我领福利'), '关注我领福利') check('读到空 → ""', vp.clean_account_name('编辑资料\n抖音号：123'), '') check('None → ""', vp.clean_account_name(None), '') check('首尾空白去掉', vp.clean_account_name(' 小王 '), '小王') print('=== 2. 临时库里造两条计划（小王 / 小李）===') with app.app_context(): def mk(name, date, seq, status=vp.ST_READY, title="标题"): row = models.VideoPlan(id=name + str(seq) + date[-2:], nickname=name, phone="", device_name="", douyin_id="", serial="", release_date=date, seq=seq, seq_auto=False, title=title, body="", video_file=f"2026-10/x{seq}.mp4", video_name="x.mp4", status=status, created_at=vp._now(), updated_at=vp._now()) models.db.session.add(row) mk('小王', '2026-09-12', 1) mk('小王', '2026-09-13', 2) # 同一个号，后天那条 mk('小李', '2026-09-12', 1) mk('小张', '2026-09-20', 1) # 未来：不该被取到 mk('小赵', '2026-09-10', 1, status=vp.ST_DONE) # 已发：不该被取到 models.db.session.commit() print('=== 3. 按名字取计划 ===') check('小王 在 09-12 有一条', [p['seq'] for p in vp.take_for_name(['小王'], '2026-09-12')], [1]) check('取 09-13 时，过期的那条先发', [p['seq'] for p in vp.take_for_name(['小王'], '2026-09-13')], [1]) check('小李 只认自己的', [p['nickname'] for p in vp.take_for_name(['小李'], '2026-09-12')], ['小李']) check('没有的名字取不到', vp.take_for_name(['不存在'], '2026-09-12'), []) check('名字为空不取（绝不猜）', vp.take_for_name([], '2026-09-12'), []) check('未来的不发', vp.take_for_name(['小张'], '2026-09-12'), []) check('已发的（done）不再入队', vp.take_for_name(['小赵'], '2026-09-12'), []) check('peek 只看不占位', vp.peek_for_name(['小王'], '2026-09-12')['seq'], 1) check('peek 不存在的名字 → None', vp.peek_for_name(['不存在'], '2026-09-12'), None) print('=== 4. pending_names / pending_count（循环块的数据源）===') # 同日期同编号时按名字的**码点**定序（只要确定就行，不追求拼音序）：李 U+674E < 王 U+738B check('有待发内容的名字（按最早日期，同日期按名字定序）', [(x['name'], x['release_date'], x['seq']) for x in vp.pending_names('2026-09-13')], [('小李', '2026-09-12', 1), ('小王', '2026-09-12', 1)]) check('只看指定的名字', [x['name'] for x in vp.pending_names('2026-09-13', names=['小王'])], ['小王']) check('小王 待发 2 条', vp.pending_count('小王', '2026-09-13'), 2) check('小李 待发 1 条', vp.pending_count('小李', '2026-09-13'), 1) check('已完成的名字不在列表里', [x['name'] for x in vp.pending_names('2026-09-12', names=['小赵'])], []) check('未来日期还没到 → 今天不列', vp.pending_count('小张', '2026-09-12'), 0) print('=== 5. 读名字步骤（假设备）===') ctx = {"job_id": "t1", "job_name": "测试"} w = gt.GenericStepsWorker("FAKE01", {}, ctx=ctx) el_raw = '小王\n编辑资料\n抖音号：35377983067' r = w._exec_read_account_name(StubD(el_raw), {"selector_type": "xpath", "selector_value": "//x", "timeout": 1}) check('读到名字 → True', r, True) check('_vars.account_name', w._vars.get('account_name'), '小王') check('_vars.name_selector 记下了', w._vars.get('name_selector'), ('xpath', '//x')) w2 = gt.GenericStepsWorker("FAKE02", {}, ctx=ctx) r2 = w2._exec_read_account_name(StubD('编辑资料\n抖音号：123'), {"selector_type": "xpath", "selector_value": "//x", "timeout": 1}) check('读不到（on_missing=fail）→ False', r2, False) check('读不到时不写 account_name', w2._vars.get('account_name'), None) w3 = gt.GenericStepsWorker("FAKE03", {}, ctx=ctx) r3 = w3._exec_read_account_name(StubD('编辑资料'), {"selector_type": "xpath", "selector_value": "//x", "on_missing": "skip"}) check('on_missing=skip → None（不失败）', r3, None) w4 = gt.GenericStepsWorker("FAKE04", {}, ctx=ctx) r4 = w4._exec_read_account_name(StubD('小王'), {"selector_type": "xpath", "selector_value": ""}) check('选择器为空 → False（并提示去抓元素）', r4, False) print('=== 6. _release_scope：这次发的是哪个号 ===') check('默认按名字（读到谁就是谁）', w._release_scope({}), ('name', ['小王'], '名字『小王』')) check('没有名字 → 空值（跳过，不猜）', w2._release_scope({}), ('name', [], '没读到本机账号名字（前面要有「读取本机账号名字」，' '或由循环块设目标名字）→ 跳过')) check('account=phone 走老路径', w._release_scope({'account': 'phone', 'account_phone': '13300000000'}), ('phone', ['13300000000'], '指定手机号 13300000000')) check('cycle 设的目标名字优先', (w._vars.__setitem__('target_name', '小李'), w._release_scope({}))[1], ('name', ['小李'], '名字『小李』')) w._vars.pop('target_name') print('=== 7. 条件判断候选值 ===') txt, note = w._resolve_cmp_value({'cmp_source': 'account_target'}, '') check('account_target → 候选=名字', txt, '小王') ok('说明里写了名字', '小王' in note, note) txt2, _ = w2._resolve_cmp_value({'cmp_source': 'account_target'}, '') check('读不到名字 → 候选为空（永远不命中 → 不发）', txt2, '') txt3, _ = w2._resolve_cmp_value({'cmp_source': 'account_target', 'cmp_value': '手填的'}, '手填的') check('读不到名字时手填值仍然生效', txt3, '手填的') print('=== 8. 推送前核对（假设备，push_one 打桩）===') calls = [] orig = flow.push_one def fake_push(serial, plan, opts, log=None): calls.append((serial, plan['id'])) return True, "ok" flow.push_one = fake_push try: # 8a 屏幕上不是我们要发的人 → 不推 w._vars['account_name'] = '小王' w._vars['name_selector'] = ('xpath', '//x') bad = w._exec_push_release(StubD('小李'), {'account': 'name', 'date_source': 'fixed', 'fixed_date': '2026-09-12'}) check('核对不过 → 不推（返回 None）', bad, None) check('push_one 一次都没调', calls, []) # 8b 屏幕上就是本人 → 推 good = w._exec_push_release(StubD('小王'), {'account': 'name', 'date_source': 'fixed', 'fixed_date': '2026-09-12'}) check('核对通过 → 推成功', good, True) check('push_one 被调了 1 次', len(calls), 1) check('推的是小王那条', calls[0][1], '小王112') # 8c 读不到名字 → 不推 calls.clear() w2._vars['name_selector'] = ('xpath', '//x') none_ = w2._exec_push_release(StubD('编辑资料'), {'date_source': 'fixed', 'fixed_date': '2026-09-12'}) check('读不到名字 → 不推', none_, None) check('push_one 没被调', calls, []) finally: flow.push_one = orig print() shutil.rmtree(TMP, ignore_errors=True) if fails: print('❌ 失败 %d 项：%s' % (len(fails), ' / '.join(fails))) sys.exit(1) print('✅ 全部通过（临时库已清理）')