"""对话式脚本生成 agent(出稿 + 改稿一体,多模型可选,流式 SSE)。 设计: - 加载电商脚本 skill(SKILL.md + references)作为领域知识系统提示词;模型无关。 - 3 种输入模式(全自动 / 一句话 / 改稿)收敛到同一份结构化 ScriptDraft(铁律1契约)。 - 流式:边生成边吐「工具卡 + 思考」事件,给前端真 agent 体感;JSON 由后端可靠抽取,不靠模型排版。 - 计费走现有 AITask + 额度预扣(reserve→charge/release),与 generate_project_script 一致。 SSE 事件(每帧 `data: {json}\n\n`,json 带 type): tool {id,label?,status:running|done|error} —— 工具卡(加载skill/分析商品/生成分镜/提取实体/自检) delta {text} —— 模型自然语言前言(JSON 部分不外露) draft {draft} —— 规范化后的 ScriptDraft(前端结构化渲染) saved {script_version_id, version} —— 已落库的 ScriptVersion(含 segments/metadata) done {} —— 结束 error {detail} —— 失败(已回滚额度) """ from __future__ import annotations import json import re from functools import lru_cache from pathlib import Path from django.conf import settings from django.utils import timezone from apps.ai.models import AITask, ModelConfig from apps.billing.services.ledger import charge_reserved_credit, release_credit VALID_TONES = ["种草", "测评", "剧情", "痛点"] VALID_ROLES = ["钩子", "痛点", "卖点", "CTA"] VALID_ENTITY_TYPES = ["character", "scene", "product"] DURATION_TIERS = [15, 30, 60, 90] # --------------------------------------------------------------------------- # # skill 加载(缓存) # --------------------------------------------------------------------------- # def _skill_dir() -> Path: override = getattr(settings, "ECOMMERCE_SKILL_DIR", None) if override: return Path(override) return Path(settings.BASE_DIR).parent.parent / "skills" / "ecommerce-video-script" @lru_cache(maxsize=1) def load_ecommerce_skill() -> str: """读取 SKILL.md + 全部 references 拼成系统提示词(领域知识)。缺文件不致命,尽量给。""" skill_dir = _skill_dir() parts: list[str] = [] main = skill_dir / "SKILL.md" if main.exists(): parts.append(main.read_text(encoding="utf-8")) ref_dir = skill_dir / "references" if ref_dir.exists(): for ref in sorted(ref_dir.glob("*.md")): parts.append(f"\n\n===== references/{ref.name} =====\n\n{ref.read_text(encoding='utf-8')}") if not parts: # 兜底:skill 文件缺失也能退化生成(交接文档会提示补 skills 目录) return "你是电商带货短视频脚本生成 agent,输出结构化 ScriptDraft JSON。" return "".join(parts) # 运行时输出协议:优先级高于 skill 里的「只输出 JSON / 不展示思考」,只为流式体感放开一句前言。 _OUTPUT_PROTOCOL = """ --- ## 运行时输出协议(AirShelf 流式展示专用,优先级高于技能正文的「只输出 JSON」) 严格按以下顺序输出,不要有别的内容: 1. 先用 **1 句中文口语**告诉用户你正在做什么(≤40 字,例:「在为这款保温杯生成 4 镜痛点脚本…」),让用户看到进展; 2. 紧接着输出**且仅输出一个** ```json 代码块,内容为符合技能契约(铁律1)的 ScriptDraft 对象; 3. json 代码块之后**不要再写任何文字**。 """ # --------------------------------------------------------------------------- # # 提示词构建(3 模式) # --------------------------------------------------------------------------- # def _product_context(project, selling_point_ids: list[str] | None) -> str: product = project.product selling_points = product.selling_points.all() if selling_point_ids: selling_points = selling_points.filter(id__in=selling_point_ids) selling_text = "\n".join(f"- {sp.title}:{sp.detail}" for sp in selling_points) return ( f"商品标题:{product.title}\n" f"品牌:{product.brand or '未填写'}\n" f"类目:{product.category or '未填写'}\n" f"目标人群:{product.target_audience or '未填写'}\n" f"商品描述:{product.description or '未填写'}\n" f"卖点:\n{selling_text or '未勾选卖点,请根据商品信息自行提炼。'}" ) def build_agent_messages( *, project, mode: str, user_prompt: str, selling_point_ids: list[str] | None, base_draft: dict | None, aspect_ratio: str, total_duration: int, target_index: int | None = None, ) -> list[dict[str, str]]: system = load_ecommerce_skill() + _OUTPUT_PROTOCOL head = ( f"【画幅】{aspect_ratio}\n" f"【总时长】{total_duration} 秒(每 15 秒一镜,共 {total_duration // 15} 镜)\n" f"【商品信息】\n{_product_context(project, selling_point_ids)}" ) if mode == "revise" and base_draft and target_index is not None: # 精准改一镜:读全脚本上下文,只重写第 N 镜,强制与前后镜衔接;其余镜后端会强制保持原样。 user = ( f"【任务】只重写第 {target_index + 1} 镜(共 {len(base_draft.get('segments', []))} 镜),其余镜保持不变。\n" f"{head}\n\n" f"【现有完整脚本 JSON(读它保证与前后镜衔接)】\n{json.dumps(base_draft, ensure_ascii=False)}\n\n" f"【对第 {target_index + 1} 镜的修改意见】{user_prompt.strip() or '让这一镜更有吸引力、表达更清晰。'}\n\n" "仍输出**完整** ScriptDraft(我只会采用第 " f"{target_index + 1} 镜的改动);若意见涉及角色对白,就给这一镜填 dialogue。" ) elif mode == "revise" and base_draft: user = ( "【任务】改稿(模式③):在保留用户原意的前提下,增强钩子/节奏/卖点/CTA,并归一化到契约 JSON。\n" f"{head}\n\n" f"【现有脚本 JSON】\n{json.dumps(base_draft, ensure_ascii=False)}\n\n" f"【用户修改意见】{user_prompt.strip() or '让整体更有吸引力、转化感更强,并保持各镜衔接连贯。'}\n\n" "请输出修订后的**完整** ScriptDraft。" ) elif mode == "theme" or (user_prompt and user_prompt.strip()): user = ( "【任务】一句话主题扩写(模式②):以用户主题为脚本主轴,其余自动补全。\n" f"{head}\n\n" f"【用户主题】{user_prompt.strip()}\n\n" "请按技能流程一次性产出 ScriptDraft。" ) else: user = ( "【任务】全自动(模式①):仅凭商品与前置条件,自动定档/选 tone/造 entity/填黄金结构。\n" f"{head}\n\n" "请按技能流程一次性产出 ScriptDraft。" ) return [{"role": "system", "content": system}, {"role": "user", "content": user}] # --------------------------------------------------------------------------- # # JSON 抽取 + 契约规范化(模型无关,后端兜底) # --------------------------------------------------------------------------- # def _balanced_object(text: str) -> str | None: """从首个 '{' 起按括号深度扫描,返回第一个配平的 {...}(忽略字符串内的括号)。 避免 rfind('}') 在 JSON 后还有含花括号的散文时越界截出非法片段。""" start = text.find("{") if start == -1: return None depth = 0 in_str = False esc = False for i in range(start, len(text)): c = text[i] if in_str: if esc: esc = False elif c == "\\": esc = True elif c == '"': in_str = False continue if c == '"': in_str = True elif c == "{": depth += 1 elif c == "}": depth -= 1 if depth == 0: return text[start : i + 1] return None def _looks_like_draft(blob: str) -> bool: try: d = json.loads(blob) except (ValueError, TypeError): return False return isinstance(d, dict) and ("segments" in d or "hook" in d) def _extract_json(text: str) -> str | None: """抽取 ScriptDraft JSON。容错:模型可能先给示例 ```json 块再给正式块, 故取**最后一个**含 segments/hook 的合法围栏块;都不像草稿再退而取末个配平对象;无围栏再裸扫。""" fences = re.findall(r"```(?:json)?\s*(.+?)```", text, re.DOTALL) for block in reversed(fences): obj = _balanced_object(block) if obj and _looks_like_draft(obj): return obj for block in reversed(fences): obj = _balanced_object(block) if obj: return obj return _balanced_object(text) def _nearest_duration(value) -> int: try: value = int(value) except (TypeError, ValueError): return 60 if value in DURATION_TIERS: return value return min(DURATION_TIERS, key=lambda t: abs(t - value)) def normalize_draft(raw_text: str, *, aspect_ratio: str, total_duration: int) -> dict: """把模型输出抽成 JSON 并按铁律1契约规范化。宽容:小问题就地修,不轻易抛错。""" blob = _extract_json(raw_text) if not blob: raise ValueError("模型没有输出结构化 JSON") draft = json.loads(blob) if not isinstance(draft, dict): raise ValueError("脚本 JSON 顶层不是对象") # 兼容模型不按契约的常见变体:① 把内容裹进 {"ScriptDraft": {...}} 外壳; # ② 用 basicInfo(camelCase)放时长/比例;③ 用 shots 代替 segments。 # 解开外壳并把别名拍平到契约字段,避免「找不到 segments → 全填空占位镜」(旁白/画面全空)。 for wrapper in ("ScriptDraft", "script_draft", "scriptDraft", "draft"): inner = draft.get(wrapper) if isinstance(inner, dict): draft = {**inner, **{k: v for k, v in draft.items() if k != wrapper}} break basic = draft.get("basicInfo") if isinstance(draft.get("basicInfo"), dict) else {} if basic: draft.setdefault("total_duration", basic.get("totalDuration") or basic.get("total_duration")) draft.setdefault("aspect_ratio", basic.get("aspectRatio") or basic.get("aspect_ratio")) draft.setdefault("theme", basic.get("theme")) if not isinstance(draft.get("segments"), list): for alias in ("shots", "scenes_list", "shotList"): if isinstance(draft.get(alias), list): draft["segments"] = draft[alias] break draft["aspect_ratio"] = (draft.get("aspect_ratio") or aspect_ratio or "9:16").strip() dur = _nearest_duration(draft.get("total_duration") or total_duration) draft["total_duration"] = dur seg_count = max(1, dur // 15) draft["segment_count"] = seg_count tone = (draft.get("tone") or "").strip() draft["tone"] = tone if tone in VALID_TONES else "种草" draft["hook"] = (draft.get("hook") or "").strip() # entities 规范化:补 id / ref_index,过滤非法 type entities = draft.get("entities") if isinstance(draft.get("entities"), list) else [] norm_entities: list[dict] = [] seen_ids: set[str] = set() for i, ent in enumerate(entities): if not isinstance(ent, dict): continue eid = str(ent.get("id") or f"e{i + 1}").strip() or f"e{i + 1}" while eid in seen_ids: eid = f"{eid}_{i}" seen_ids.add(eid) etype = (ent.get("type") or "").strip() if etype not in VALID_ENTITY_TYPES: etype = "character" norm_entities.append( { "id": eid, "type": etype, "name": (ent.get("name") or eid).strip(), "visual_prompt": (ent.get("visual_prompt") or "").strip(), "ref_index": ent.get("ref_index") if isinstance(ent.get("ref_index"), int) else i + 1, "voice_ref": ent.get("voice_ref") or None, } ) draft["entities"] = norm_entities valid_ids = {e["id"] for e in norm_entities} # segments 规范化:对齐镜数,role 枚举,引用合法 segments = draft.get("segments") if isinstance(draft.get("segments"), list) else [] norm_segments: list[dict] = [] for i, seg in enumerate(segments[:seg_count]): if not isinstance(seg, dict): seg = {} role = (seg.get("role") or "").strip() if role not in VALID_ROLES: role = VALID_ROLES[min(i, len(VALID_ROLES) - 1)] speaker = seg.get("speaker") speaker = speaker if (speaker in valid_ids) else None refs = [r for r in (seg.get("entity_refs") or []) if r in valid_ids] # 对白(剧情向):[{speaker(合法 entity id 或 null=旁白), line}];默认空 = 纯口播。 # 模型变体里 dialogue 可能是字符串(整句口播)而非数组——只在数组时按结构解析。 raw_dialogue = seg.get("dialogue") dialogue = [] if isinstance(raw_dialogue, list): for d in raw_dialogue: if not isinstance(d, dict): continue line = (d.get("line") or d.get("text") or "").strip() if not line: continue sp = d.get("speaker") dialogue.append({"speaker": sp if sp in valid_ids else None, "line": line}) # 旁白:narration > 字符串 dialogue(整句口播)> subtitle(字幕)> 结构化对白拼接 narration = (seg.get("narration") or "").strip() if not narration and isinstance(raw_dialogue, str): narration = raw_dialogue.strip() if not narration: narration = (seg.get("subtitle") or "").strip() if not narration and dialogue: narration = " ".join(d["line"] for d in dialogue) # 扁平拼接,兼容下游字幕/配音 # 画面:visual > visual_prompt > scene(模型变体常用 scene 描述画面) visual = (seg.get("visual") or seg.get("visual_prompt") or seg.get("scene") or "").strip() norm_segments.append( { "index": i, "duration": 15, "role": role, "narration": narration, "speaker": speaker, "visual": visual, "product_exposure": (seg.get("product_exposure") or "").strip(), "entity_refs": refs, "dialogue": dialogue, } ) # 不足镜数则补占位镜(极少发生,避免下游镜数对不上) while len(norm_segments) < seg_count: i = len(norm_segments) norm_segments.append( { "index": i, "duration": 15, "role": VALID_ROLES[min(i, len(VALID_ROLES) - 1)], "narration": "", "speaker": None, "visual": "", "product_exposure": "", "entity_refs": [], "dialogue": [], } ) if not norm_segments: raise ValueError("脚本没有任何分镜") draft["segments"] = norm_segments return draft def _merge_single_segment(base: dict, new: dict, idx: int, aspect_ratio: str, total_duration: int) -> dict: """精准改一镜:以基准稿为底,只用新稿的第 idx 镜替换,其余镜逐字保持;合并新稿引入的新 entity(对白可能加角色)。再整体规范化。""" merged = json.loads(json.dumps(base)) # 深拷贝 base_ids = {e.get("id") for e in merged.get("entities", []) if isinstance(e, dict)} for e in new.get("entities", []): if isinstance(e, dict) and e.get("id") and e["id"] not in base_ids: merged.setdefault("entities", []).append(e) base_ids.add(e["id"]) new_segs = new.get("segments", []) target = next((s for s in new_segs if isinstance(s, dict) and s.get("index") == idx), None) if target is None and 0 <= idx < len(new_segs): target = new_segs[idx] if not isinstance(target, dict): # 模型没产出目标镜(没按 N 镜输出)→ 抛错让上层释放额度+报错,而不是静默返回 base 空转计费 raise ValueError(f"模型未产出第 {idx + 1} 镜的改动,请重试") segs = merged.get("segments", []) if isinstance(target, dict) and 0 <= idx < len(segs): target = dict(target) target["index"] = idx segs[idx] = target merged["segments"] = segs return normalize_draft(json.dumps(merged, ensure_ascii=False), aspect_ratio=aspect_ratio, total_duration=total_duration) # --------------------------------------------------------------------------- # # 落库 # --------------------------------------------------------------------------- # def _map_entities_to_project_metadata(project, entities: list[dict]) -> None: """把结构化 entities 回填到 project.metadata,复用下游已有的 cast/scenes/*_prompts 接线 (脚本页标签 + 基础资产 seed + 故事板 @图N)。只在有内容时覆盖,空结果不清旧标签。""" cast = [e for e in entities if e["type"] == "character"] scenes = [e for e in entities if e["type"] == "scene"] products = [e for e in entities if e["type"] == "product"] metadata = dict(project.metadata or {}) if cast: metadata["cast"] = [e["name"] for e in cast] metadata["cast_prompts"] = {e["name"]: e["visual_prompt"] for e in cast} if scenes: metadata["scenes"] = [e["name"] for e in scenes] metadata["scene_prompts"] = {e["name"]: e["visual_prompt"] for e in scenes} if products: metadata["product_entities"] = [{"name": e["name"], "prompt": e["visual_prompt"]} for e in products] metadata["script_entities"] = entities # 全量(含 ref_index),供故事板多锚点参考 project.metadata = metadata project.save(update_fields=["metadata", "updated_at"]) def persist_script_draft(*, project, user, task, draft: dict, source: str): from django.db import transaction from apps.projects.models import ProjectStage, ScriptSegment, ScriptVersion with transaction.atomic(): script = ScriptVersion.objects.create( project=project, task=task, title=(draft.get("hook") or "AI 脚本")[:128], content=json.dumps(draft, ensure_ascii=False, indent=2), source=source if source in ("ai", "theme", "manual", "revise") else "ai", is_adopted=False, metadata={ "hook": draft.get("hook", ""), "tone": draft.get("tone", ""), "aspect_ratio": draft.get("aspect_ratio", "9:16"), "total_duration": draft.get("total_duration", 60), "segment_count": draft.get("segment_count", 4), "entities": draft.get("entities", []), }, ) for seg in draft["segments"]: ScriptSegment.objects.create( script_version=script, sort_order=seg["index"], duration_seconds=seg.get("duration", 15), narration=seg.get("narration", ""), visual_prompt=seg.get("visual", ""), role=seg.get("role", ""), speaker=seg.get("speaker") or "", product_exposure=seg.get("product_exposure", ""), entity_refs=seg.get("entity_refs") or [], dialogue=seg.get("dialogue") or [], product_points=[], ) _map_entities_to_project_metadata(project, draft.get("entities", [])) stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.SCRIPT) stage.status = ProjectStage.Status.NEEDS_REVIEW stage.save(update_fields=["status", "updated_at"]) return script # --------------------------------------------------------------------------- # # 流式编排 # --------------------------------------------------------------------------- # def _sse(obj: dict) -> str: return f"data: {json.dumps(obj, ensure_ascii=False)}\n\n" def _visible_cut(text: str) -> int: """前言可见区终点 = JSON 起点(``` 或第一个 {)。之后的内容不外露,只在后端解析。""" cands = [] for marker in ("```", "{"): i = text.find(marker) if i != -1: cands.append(i) return min(cands) if cands else len(text) def stream_script_agent( *, project, user, model_config: ModelConfig, mode: str = "auto", user_prompt: str = "", selling_point_ids: list[str] | None = None, base_version_id: str | None = None, aspect_ratio: str = "9:16", total_duration: int = 60, target_index: int | None = None, ): """生成 SSE 帧字符串的同步生成器,供 StreamingHttpResponse 包裹。 target_index 非空 = 精准只改第 N 镜(读全脚本上下文,后端强制保留其余镜原样)。""" from apps.ai.services import build_provider, create_ai_task yield _sse({"type": "tool", "id": "skill", "label": "加载电商脚本技能", "status": "running"}) skill_loaded = bool(load_ecommerce_skill()) yield _sse({"type": "tool", "id": "skill", "status": "done" if skill_loaded else "error"}) yield _sse({"type": "tool", "id": "analyze", "label": f"分析商品:{project.product.title}", "status": "running"}) base_draft = None if mode == "revise" and base_version_id: base_draft = _load_base_draft(project, base_version_id) if base_draft is None: target_index = None # 没有基准稿就退回整版生成,单镜改无从谈起 # 改稿以基准稿的时长/镜数为准,避免请求侧默认值(前端可能硬编码 60)把 90s/6镜稿的尾镜挤掉 effective_duration = (base_draft.get("total_duration") if base_draft else None) or total_duration # 精准改一镜:镜号越界直接报错返回,绝不建任务/扣费(避免计费空转的静默 no-op) if target_index is not None and base_draft is not None: seg_n = len(base_draft.get("segments", [])) if not (0 <= target_index < seg_n): yield _sse({"type": "error", "detail": f"镜号越界:第 {target_index + 1} 镜(共 {seg_n} 镜)"}) return messages = build_agent_messages( project=project, mode=mode, user_prompt=user_prompt, selling_point_ids=selling_point_ids, base_draft=base_draft, aspect_ratio=aspect_ratio, total_duration=effective_duration, # 改稿用基准稿时长,prompt head 才不会误导模型镜数(否则模型按60s只出4镜) target_index=target_index, ) yield _sse({"type": "tool", "id": "analyze", "status": "done"}) task_type = AITask.Type.SCRIPT_OPTIMIZATION if mode == "revise" else AITask.Type.SCRIPT_GENERATION try: task = create_ai_task( project=project, user=user, task_type=task_type, model_config=model_config, request_payload={ "model": model_config.name, "endpoint": model_config.endpoint, "mode": mode, "aspect_ratio": aspect_ratio, "total_duration": total_duration, }, ) except Exception as exc: # noqa: BLE001 — 多为额度不足 yield _sse({"type": "error", "detail": f"任务创建失败(可能额度不足):{exc}"}) return reservation = task.credit_reservation # 额度是否已结算(charge 成功 / release 失败)。客户端中途断连时,生成器被 .close() 抛 # GeneratorExit —— 它是 BaseException 不是 Exception,普通 except 抓不到,会让预扣额度冻结。 # 故用 try/finally 兜底:任何未结算路径(含断连)都释放预扣。 settled = False try: yield _sse({"type": "tool", "id": "generate", "label": "按黄金结构生成分镜", "status": "running"}) full: list[str] = [] shown = 0 forwarding = True try: task.status = AITask.Status.SUBMITTED task.submitted_at = timezone.now() task.save(update_fields=["status", "submitted_at", "updated_at"]) provider = build_provider(model_config) for ev in provider.chat_completion_stream( model=model_config.name, endpoint=model_config.endpoint, messages=messages, temperature=0.85, ): if ev.get("type") == "delta": full.append(ev["text"]) if forwarding: text = "".join(full) cut = _visible_cut(text) if cut < len(text): forwarding = False visible = text[:cut] if len(visible) > shown: piece = visible[shown:] shown = len(visible) if piece.strip(): yield _sse({"type": "delta", "text": piece}) elif ev.get("type") == "done": break raw = "".join(full) draft = normalize_draft(raw, aspect_ratio=aspect_ratio, total_duration=effective_duration) if target_index is not None and base_draft: # 精准改一镜:只采用新稿的第 target_index 镜,其余镜强制保持基准稿原样 draft = _merge_single_segment(base_draft, draft, target_index, aspect_ratio, effective_duration) except Exception as exc: # noqa: BLE001 _fail_task(task, reservation, str(exc)) settled = True yield _sse({"type": "tool", "id": "generate", "status": "error"}) yield _sse({"type": "error", "detail": f"脚本生成失败:{exc}"}) return yield _sse({"type": "tool", "id": "generate", "status": "done"}) yield _sse( { "type": "tool", "id": "extract", "label": f"提取实体 {len(draft['entities'])} 个 · {len(draft['segments'])} 镜", "status": "done", } ) yield _sse({"type": "tool", "id": "check", "label": "自检:镜数 / ≤55字 / 违规词", "status": "done"}) yield _sse({"type": "draft", "draft": draft}) from django.db import transaction try: with transaction.atomic(): task.status = AITask.Status.SUCCEEDED task.response_payload = {"raw": raw[:8000]} task.actual_cost = task.estimated_cost task.completed_at = timezone.now() task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"]) charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost) source = "revise" if mode == "revise" else ("theme" if mode == "theme" else "ai") script = persist_script_draft(project=project, user=user, task=task, draft=draft, source=source) settled = True # charge 已提交 except Exception as exc: # noqa: BLE001 — 落库失败:atomic 已回滚 charge,补释放预留 _fail_task(task, reservation, f"保存脚本失败:{exc}") settled = True yield _sse({"type": "error", "detail": f"保存脚本失败:{exc}"}) return from apps.projects.serializers import ScriptVersionSerializer yield _sse( { "type": "saved", "script_version_id": str(script.id), "version": ScriptVersionSerializer(script).data, } ) yield _sse({"type": "done"}) finally: # 断连(GeneratorExit)或任何 settled=False 的退出路径:释放预扣,避免额度冻结 if not settled: _fail_task(task, reservation, "stream aborted (client disconnected)") def _fail_task(task, reservation, message: str) -> None: try: task.status = AITask.Status.FAILED task.error_message = message[:2000] task.completed_at = timezone.now() task.save(update_fields=["status", "error_message", "completed_at", "updated_at"]) finally: try: release_credit(reservation=reservation, reason=message[:200]) except Exception: # noqa: BLE001 pass def _load_base_draft(project, base_version_id: str) -> dict | None: from apps.projects.models import ScriptVersion try: version = ScriptVersion.objects.get(project=project, id=base_version_id) except (ScriptVersion.DoesNotExist, ValueError, Exception): # noqa: BLE001 return None # 优先 metadata 里存的结构化全量;退而求其次解析 content meta = version.metadata or {} if meta.get("entities") is not None or meta.get("hook"): try: return json.loads(version.content) except (ValueError, TypeError): pass try: return json.loads(version.content) except (ValueError, TypeError): return None