Merge pull request #202 from zhouxiaoka/codex/analysis-routing

feat: enforce explicit analysis routing in unified import

backend/api/v1/studio.py

@@ -47,7 +47,16 @@ def title_preset_thumbnail(style: Literal['comic', 'neon', 'arena', 'editorial',
 def capabilities():
     return {'visual_analysis': intelligence.ready(), 'visual_model': intelligence.visual_config()[2], 'languages': ['source', 'zh', 'en', 'ja']}
 
-from backend.services.studio import vision_settings
+from backend.services.studio import vision_settings, analysis_preferences
+
+@router.get('/analysis-preferences')
+def get_analysis_preferences():
+    return call(analysis_preferences.load)
+
+@router.put('/analysis-preferences')
+def save_analysis_preferences(body: analysis_preferences.AnalysisPreferences):
+    return call(analysis_preferences.save, body)
+
 
 @router.get('/vision-settings')
 def get_vision_settings():

backend/services/studio/jobs.py

@@ -189,11 +189,21 @@ def confirm_project(project_id, body):
         plan = state.get('plan') or {}
         if plan.get('id') != body.plan_id or (state.get('analysis') or {}).get('status') != 'awaiting_confirmation':
             raise store.ConflictError('方案已变化或任务已开始，请刷新后确认')
+        route = body.analysis_mode or plan.get('recommended_analysis', 'subtitle')
+        if route == 'subtitle' and any(goal != 'content' for goal in body.goals):
+            raise ValueError('字幕链路目前支持内容切片；视觉高光与推广分析需要显式选择视觉模式，原素材已保留')
+        if route == 'visual' and body.goals == ['content']:
+            raise ValueError('内容切片当前使用字幕分析，请选择字幕模式后确认')
+        if route == 'visual':
+            from backend.services.studio import intelligence
+            if not intelligence.ready():
+                raise ValueError('视觉模型不可用，请配置后重新确认；原素材已保留')
         previous = deepcopy(state)
+        plan['confirmed_analysis'] = route
         plan['selected_goals'] = body.goals
         plan['confirmed_at'] = store.now()
         # A confirmation override is persisted separately from the AI recommendation.
-        plan['confirmed_preferences'] = {**plan['preferences'], 'goal':body.goals[0], **body.model_dump(exclude={'plan_id','goals'}, exclude_none=True)}
+        plan['confirmed_preferences'] = {**plan['preferences'], 'goal':body.goals[0], **body.model_dump(exclude={'plan_id','goals','analysis_mode'}, exclude_none=True)}
         state['analysis'] = {'status':'running', 'phase':'production', 'message':'开始制作所选内容', 'instance':store.INSTANCE, 'created_at':store.now()}
         store.write(project_id, state)
         try:
@@ -227,6 +237,8 @@ def _produce_selected(project_id, plan):
                     run_content(project_id, video)
                     mark_project(project_id, 'processing')
                     continue
+                if plan.get('confirmed_analysis', 'subtitle') != 'visual':
+                    raise ValueError('此确认未授权视觉分析；请重新选择处理方式')
                 if not intelligence.ready():
                     raise ValueError('请先在设置中配置视觉理解模型')
                 if visual_error is not None:

backend/services/studio/models.py

@@ -99,6 +99,7 @@ class ImportOptions(BaseModel):
 
 class ConfirmPlan(BaseModel):
     model_config = ConfigDict(extra='forbid')
+    analysis_mode: Literal['subtitle', 'visual'] | None = None
     plan_id: str = Field(min_length=1, max_length=100)
     goals: list[Goal] = Field(min_length=1, max_length=3)
     language: Language | None = None

backend/services/studio/planning.py

@@ -3,7 +3,7 @@ from pathlib import Path
 import tempfile
 from typing import Literal
 from pydantic import BaseModel, Field
-from backend.services.studio import intelligence
+from backend.services.studio import intelligence, analysis_preferences
 from backend.services.studio.models import ImportOptions, Preferences
 
 
@@ -22,13 +22,19 @@ def recommend(video: Path, options: ImportOptions):
     duration = info.get('duration', 0)
     if duration < 1 or duration > 7200:
         raise ValueError('智能制作目前支持 1 秒至 2 小时的素材')
+    consent = analysis_preferences.load()
+    configured = intelligence.ready()
     mode = 'manual'
     diagnostics = None
     if options.goal != 'auto':
         result = Recommendation(content_type='other', goal=options.goal,
             reason='按你指定的制作方式处理，其他未指定选项使用推荐设置。', confidence=1,
             aspect='portrait' if options.goal == 'promo' else 'original')
-    elif not intelligence.ready():
+    elif consent.analysis_mode == 'subtitle' or (consent.analysis_mode == 'auto' and not consent.allow_visual_screening) or (configured and not analysis_preferences.visual_screening_allowed(consent, vision_configured=configured)):
+        mode = 'local'
+        result = Recommendation(content_type='other', goal='content', confidence=0, suggested_goals=['content'],
+            reason='按字幕链路处理，未调用视觉初筛；正式制作时使用已有字幕或转写，尚未验证语音内容。')
+    elif not configured:
         mode = 'fallback'
         result = Recommendation(content_type='other', goal='content',
             reason='尚未配置视觉模型，无法自动判断；请手动选择制作类型，或在设置中配置后重新识别。', confidence=0, suggested_goals=[])
@@ -64,5 +70,6 @@ def recommend(video: Path, options: ImportOptions):
     prefs = Preferences(goal=result.goal, language=options.language,
         aspect=options.aspect or result.aspect, duration=options.duration or result.duration)
     suggested = list(dict.fromkeys(result.suggested_goals if result.suggested_goals is not None else (["highlight", "promo"] if mode == 'ai' and result.content_type == 'gameplay' else [result.goal])))
-    return {**({'diagnostics': diagnostics} if diagnostics else {}), 'mode': mode, 'source_duration': duration, **result.model_dump(), 'suggested_goals': suggested, 'preferences': prefs.model_dump(),
+    route = 'visual' if result.goal != 'content' and (consent.analysis_mode == 'visual' or (consent.analysis_mode == 'auto' and mode == 'ai' and result.goal != 'content')) else 'subtitle'
+    return {'analysis_preferences': consent.model_dump(), 'recommended_analysis': route, **({'diagnostics': diagnostics} if diagnostics else {}), 'mode': mode, 'source_duration': duration, **result.model_dump(), 'suggested_goals': suggested, 'preferences': prefs.model_dump(),
             'overrides': options.model_dump(exclude_none=True)}

backend/tests/test_analysis_routing.py

@@ -0,0 +1,53 @@
+"""Cost boundaries at the real import/confirmation endpoints, without paid calls."""
+import pytest
+from backend.tests.test_studio import root, source, client
+from backend.services.studio import analysis_preferences as ap, intelligence, jobs, planning, store
+from backend.services.studio.models import ImportOptions
+
+class Immediate:
+    def submit(self, fn, *args):
+        fn(*args)
+
+@pytest.mark.parametrize('mode', ['subtitle', 'auto', 'visual'])
+@pytest.mark.parametrize('configured', [False, True])
+@pytest.mark.parametrize('subtitles', [False, True])
+def test_unapproved_screening_never_calls_vision(source, monkeypatch, mode, configured, subtitles):
+    monkeypatch.setattr(ap, 'load', lambda: ap.AnalysisPreferences(analysis_mode=mode))
+    monkeypatch.setattr(intelligence, 'ready', lambda: configured)
+    monkeypatch.setattr(intelligence, 'vision_call', lambda *a, **k: pytest.fail('unapproved vision call'))
+    if subtitles:
+        (source.parent / 'input.srt').write_text('1\n00:00:00,000 --> 00:00:01,000\nHello\n')
+    result = planning.recommend(source, ImportOptions())
+    assert result['mode'] != 'ai'
+    assert result['analysis_preferences']['allow_visual_screening'] is False
+
+
+def test_subtitle_confirm_uses_only_old_pipeline_and_freezes_route(client, source, monkeypatch):
+    monkeypatch.setattr(ap, 'load', lambda: ap.AnalysisPreferences())
+    monkeypatch.setattr(intelligence, 'ready', lambda: True)
+    monkeypatch.setattr(intelligence, 'vision_call', lambda *a, **k: pytest.fail('vision forbidden'))
+    monkeypatch.setattr(jobs, 'analyze', lambda *a, **k: pytest.fail('visual analysis forbidden'))
+    monkeypatch.setattr(jobs, 'executor', Immediate())
+    calls = []
+    monkeypatch.setattr(jobs, 'run_content', lambda *a: calls.append(a))
+    response = client.post('/studio/import', files={'video': ('source.mp4', source.read_bytes(), 'video/mp4')})
+    assert response.status_code == 200
+    pid = response.json()['project_id']
+    state = store.read(pid)
+    assert not calls and state['analysis']['status'] == 'awaiting_confirmation'
+    # Changing global settings cannot upgrade an already screened plan.
+    monkeypatch.setattr(ap, 'load', lambda: ap.AnalysisPreferences(analysis_mode='visual'))
+    rejected = client.post(f'/studio/{pid}/start', json={'plan_id':state['plan']['id'], 'goals':['highlight']})
+    assert rejected.status_code == 422
+    assert store.read(pid)['analysis']['status'] == 'awaiting_confirmation'
+    accepted = client.post(f'/studio/{pid}/start', json={'plan_id':state['plan']['id'], 'goals':['content']})
+    assert accepted.status_code == 200 and len(calls) == 1
+    assert store.read(pid)['plan']['confirmed_analysis'] == 'subtitle'
+
+
+def test_settings_api_strict_contract(client):
+    assert client.get('/studio/analysis-preferences').json()['analysis_mode'] == 'subtitle'
+    assert client.put('/studio/analysis-preferences', json={'analysis_mode':'subtitle','allow_visual_screening':True}).status_code == 422
+    assert client.put('/studio/analysis-preferences', json={'analysis_mode':'invented'}).status_code == 422
+    assert client.put('/studio/analysis-preferences', json={'analysis_mode':'auto','allow_visual_screening':True}).status_code == 200
+    assert client.get('/studio/analysis-preferences').json()['allow_visual_screening'] is True

backend/tests/test_smart_import.py

@@ -7,6 +7,12 @@ from backend.tests.test_studio import root, source, client
 from backend.services.studio import intelligence, jobs, store, planning
 from backend.services.studio.models import ImportOptions
 
+@pytest.fixture(autouse=True)
+def explicit_visual_preferences(monkeypatch):
+    # These existing tests exercise the opted-in visual workflow.
+    from backend.services.studio import analysis_preferences as ap
+    monkeypatch.setattr(ap, 'load', lambda: ap.AnalysisPreferences(analysis_mode='visual', allow_visual_screening=True))
+
 class Immediate:
     def submit(self, fn, *args): fn(*args)
 

backend/tests/test_studio.py

@@ -201,7 +201,7 @@ def test_visual_import_worker_persists_project_status(client,root,source,monkeyp
     pid=response.json()['project_id']
     state=client.get('/studio/'+pid).json()
     assert state['analysis']['status']=='awaiting_confirmation' and state['drafts']==[]
-    assert client.post('/studio/'+pid+'/start',json={'plan_id':state['plan']['id'],'goals':['highlight']}).status_code==200
+    assert client.post('/studio/'+pid+'/start',json={'plan_id':state['plan']['id'],'goals':['highlight'],'analysis_mode':'visual'}).status_code==200
     with SessionLocal() as db:
         p=db.get(Project,pid)
         assert p.status==ProjectStatus.COMPLETED

backend/tests/test_vision_failures.py

@@ -77,6 +77,9 @@ def test_refusal_and_valid_fenced_json(monkeypatch):
 
 
 def test_stage_annotation_and_quick_fallback_preserve_diagnostics(monkeypatch):
+    from backend.services.studio import analysis_preferences as ap
+    # Exercise timeout diagnostics only after explicitly authorizing paid screening.
+    monkeypatch.setattr(ap, 'load', lambda: ap.AnalysisPreferences(analysis_mode='auto', allow_visual_screening=True))
     def fail(*a,**k):raise vision.VisionRequestError('timeout','模型超时',elapsed_seconds=30)
     monkeypatch.setattr(vision,'vision_call',fail)
     with pytest.raises(vision.VisionRequestError) as caught:vision.vision_call_at('refine',[])
@@ -87,3 +90,14 @@ def test_stage_annotation_and_quick_fallback_preserve_diagnostics(monkeypatch):
     plan=planning.recommend(Path('unused'),ImportOptions())
     assert plan['mode']=='fallback' and plan['suggested_goals']==[]
     assert plan['diagnostics']=={'code':'timeout','phase':'screening','elapsed_seconds':30}
+
+
+def test_unapproved_screening_does_not_attempt_provider_or_fabricate_timeout(monkeypatch):
+    from backend.services.studio import analysis_preferences as ap
+    monkeypatch.setattr(ap, 'load', lambda: ap.AnalysisPreferences())
+    monkeypatch.setattr(vision, 'ready', lambda: True)
+    monkeypatch.setattr(vision, '_probe', lambda _: {'duration': 20})
+    monkeypatch.setattr(vision, 'vision_call', lambda *a, **k: pytest.fail('unapproved provider call'))
+    plan = planning.recommend(Path('unused'), ImportOptions())
+    assert plan['mode'] == 'local'
+    assert 'diagnostics' not in plan

docs/DUAL_ANALYSIS_ROLLOUT_1_4.md

@@ -53,3 +53,13 @@
 此项仅内部存储与纯权限判断，不暴露API或模式选择器，不改变现有导入和正式分析路由；不能据此宣称产品已经满足字幕模式零视觉调用。下一PR将接入路由、确认快照和API，再接设置与导入交互。
 
 验证：19项新增偏好测试及31项已有智能导入测试全部通过；新增文件Ruff E9/F63/F7/F82通过。测试使用隔离目录与模拟模型，不调用付费接口。
+
+## 9月26日：后端授权路由（独立草稿PR，依赖#201）
+
+`codex/analysis-routing` 接入偏好API、导入方案偏好快照、recommended_analysis和确认时confirmed_analysis。字幕模式不执行视觉初筛；无初筛许可的auto使用本地保守建议，不伪称AI识别。确认可显式校正subtitle/visual，校验失败保留待确认状态；修改全局偏好不改变已生成方案。正式视觉制作检查已确认路线。
+
+当前适配范围：内容切片仍走原字幕管线，高光/推广仍使用视觉事件管线。字幕来源的高光/推广适配、更加准确的本地证据推荐、前端模式校正和八语文案尚未完成。本PR须保持草稿，不能单独合入供用户使用；旧视觉专用端点也尚未纳入全局偏好治理，不能把统一导入测试泛化为全产品零视觉调用。
+
+已有视觉导入测试显式启用视觉偏好；新增测试覆盖3种模式×2种配置×2种字幕状态下未授权初筛调用数为0，以及导入→确认的字幕执行、配置变更不升级、非法选择保留暂存、偏好API严格校验。均不调用付费接口。
+
+验证结果：首次相关+Studio回归156通过，1个旧用例缺少显式视觉确认而失败；更新该用例后相关65项全部通过（涵盖此前64项及修正用例）。其余Studio用例未改动。变更文件关键Ruff规则通过，jobs.py的5处异常闭包F821在基线中同样存在，不宣称全文件Lint通过。