fix: 统一视频延长回退与 post_id 元数据来源

This commit is contained in:
XianYuDaXian committed 2026-03-14 18:04:59 +08:00
1 parent 4b38fda7fa
commit 8cd0e780cb
7 files changed
+612 -54

No files matched your search

+310 -10
View File
@@ -135,6 +135,41 @@ def _log_final_video_payload(
logger.info(f"Video upstream payload before send:\n{payload_text}")
def _log_raw_video_stream_event(raw_text: str) -> None:
text = str(raw_text or "").strip()
if not text:
return
try:
payload = orjson.loads(text)
pretty = orjson.dumps(payload, option=orjson.OPT_INDENT_2).decode("utf-8")
except Exception:
pretty = text
logger.info(f"Video upstream raw event:\n{pretty}")
def _truncate_video_stream_line(raw_text: str, limit: int = 4000) -> str:
text = str(raw_text or "")
if len(text) <= limit:
return text
return f"{text[:limit]}...(len={len(text)})"
def _log_video_stream_line(*, stage: str, raw_text: str) -> None:
text = str(raw_text or "")
if not text:
logger.info(f"Video upstream {stage}: <empty>")
return
logger.info(
"Video upstream line "
f"({stage}): {_truncate_video_stream_line(text)}"
)
def _log_video_stream_end(*, stage: str, reason: str, extra: str = "") -> None:
suffix = f", {extra}" if extra else ""
logger.info(f"Video upstream stream ended ({stage}): reason={reason}{suffix}")
def _classify_video_error(exc: Exception) -> tuple[str, str, int]:
"""将底层异常归一化为用户可读错误。"""
text = str(exc or "").lower()
@@ -175,12 +210,12 @@ async def _try_log_video_share_link(
*,
local_url: str = "",
thumbnail_url: str = "",
) -> None:
) -> str:
"""视频生成完成后尝试创建分享链接并写入元数据。"""
token_text = str(token or "").strip()
post_text = str(post_id or "").strip()
if not token_text or not post_text:
return
return ""
try:
logger.info(f"Video create-link attempt: post_id={post_text}")
async with AsyncSession() as session:
@@ -204,12 +239,47 @@ async def _try_log_video_share_link(
"Video create-link completed without shareLink: "
f"post_id={post_text}, metadata_path={metadata_path or '-'}"
)
return share_link
except Exception as e:
details = getattr(e, "details", None)
logger.warning(
"Video create-link failed: "
f"post_id={post_text}, error={e}, details={details}"
)
return ""
async def _fetch_share_page_media_urls(share_link: str) -> tuple[str, str]:
share_text = str(share_link or "").strip()
if not share_text:
return "", ""
try:
async with AsyncSession() as session:
response = await session.get(share_text, timeout=get_config("video.timeout"))
html = str(getattr(response, "text", "") or "")
if not html:
return "", ""
video_match = re.search(
r'<meta\s+property="og:video"\s+content="([^"]+)"',
html,
flags=re.IGNORECASE,
)
image_match = re.search(
r'<meta\s+property="og:image"\s+content="([^"]+)"',
html,
flags=re.IGNORECASE,
)
video_url = str(video_match.group(1) if video_match else "").strip()
image_url = str(image_match.group(1) if image_match else "").strip()
if video_url or image_url:
logger.info(
"Video share page meta resolved: "
f"share_link={share_text}, video_url={video_url or '-'}, image_url={image_url or '-'}"
)
return video_url, image_url
except Exception as e:
logger.warning(f"Video share page fetch failed: share_link={share_text}, error={e}")
return "", ""
class VideoService:
@@ -1257,19 +1327,13 @@ class VideoService:
)
# Process response.
idle_timeout_override = None
if extend_post_id and video_extension_start_time is not None:
idle_timeout_override = (
get_config("video.extension_stream_timeout")
or get_config("video.stream_timeout")
)
if is_stream:
processor = VideoStreamProcessor(
model,
token,
show_think,
upscale_on_finish=should_upscale,
idle_timeout_override=idle_timeout_override,
idle_timeout_override=None,
)
return wrap_stream_with_usage(
processor.process(response), token_mgr, token, model
@@ -1279,7 +1343,7 @@ class VideoService:
model,
token,
upscale_on_finish=should_upscale,
idle_timeout_override=idle_timeout_override,
idle_timeout_override=None,
).process(response)
try:
model_info = ModelService.get(model)
@@ -1376,6 +1440,108 @@ class VideoStreamProcessor(BaseProcessor):
logger.warning(f"Video upscale failed: {e}")
return video_url
async def _try_render_from_post(
self,
*,
post_id: str,
progress: int = 0,
thumbnail_url: str = "",
) -> str:
post_text = str(post_id or "").strip()
if not post_text or not self.token:
return ""
logger.info(
"Video stream high-progress fallback: "
f"try create-link/media-post lookup, post_id={post_text}, progress={progress}"
)
share_link = await _try_log_video_share_link(
self.token,
post_text,
thumbnail_url=thumbnail_url,
)
if share_link:
share_video_url, share_thumb_url = await _fetch_share_page_media_urls(share_link)
if share_video_url:
render_started_at = time.perf_counter()
logger.info(
"Video stream high-progress share-link render started: "
f"post_id={post_text}, media_url={share_video_url}, thumbnail_url={share_thumb_url or thumbnail_url or '-'}"
)
rendered = await self._get_dl().render_video(
share_video_url,
self.token,
share_thumb_url or thumbnail_url,
)
render_duration_ms = (time.perf_counter() - render_started_at) * 1000
logger.info(
"Video stream high-progress share-link render completed: "
f"post_id={post_text}, duration_ms={render_duration_ms:.2f}, rendered={rendered}"
)
return rendered
post = await _fetch_media_post_info(self.token, post_text)
media_url = _normalize_assets_url(str(post.get("mediaUrl") or "").strip())
thumb_url = _normalize_assets_url(
str(post.get("thumbnailImageUrl") or thumbnail_url or "").strip()
)
if not media_url:
logger.info(
"Video stream high-progress fallback miss: "
f"post_id={post_text}, progress={progress}"
)
return ""
render_started_at = time.perf_counter()
logger.info(
"Video stream high-progress fallback render started: "
f"post_id={post_text}, media_url={media_url}, thumbnail_url={thumb_url or '-'}"
)
rendered = await self._get_dl().render_video(media_url, self.token, thumb_url)
render_duration_ms = (time.perf_counter() - render_started_at) * 1000
logger.info(
"Video stream high-progress fallback render completed: "
f"post_id={post_text}, duration_ms={render_duration_ms:.2f}, rendered={rendered}"
)
return rendered
async def _try_render_from_post_with_retry(
self,
*,
post_id: str,
progress: int = 0,
thumbnail_url: str = "",
retry_delay_seconds: int = 30,
) -> str:
rendered = await self._try_render_from_post(
post_id=post_id,
progress=progress,
thumbnail_url=thumbnail_url,
)
if rendered:
return rendered
logger.info(
"Video stream high-progress fallback retry scheduled: "
f"post_id={post_id}, progress={progress}, delay={retry_delay_seconds}s"
)
await asyncio.sleep(max(1, int(retry_delay_seconds)))
return await self._try_render_from_post(
post_id=post_id,
progress=progress,
thumbnail_url=thumbnail_url,
)
async def _wait_before_eof_fallback(
self,
*,
post_id: str,
progress: int,
wait_seconds: int = 60,
) -> None:
delay = max(1, int(wait_seconds))
logger.info(
"Video stream EOF fallback wait scheduled: "
f"post_id={post_id}, progress={progress}, delay={delay}s"
)
await asyncio.sleep(delay)
def _sse(self, content: str = "", role: str = None, finish: str = None) -> str:
"""Build SSE response."""
delta = {}
@@ -1401,16 +1567,32 @@ class VideoStreamProcessor(BaseProcessor):
) -> AsyncGenerator[str, None]:
"""Process video stream response."""
idle_timeout = self.idle_timeout_override or get_config("video.stream_timeout")
latest_progress = 0
latest_video_post_id = ""
latest_thumbnail_url = ""
final_rendered = False
try:
async for line in _with_idle_timeout(response, idle_timeout, self.model):
raw_line = (
line.decode("utf-8", errors="ignore")
if isinstance(line, (bytes, bytearray))
else str(line)
)
_log_video_stream_line(stage="stream/raw", raw_text=raw_line)
line = _normalize_line(line)
if not line:
_log_video_stream_line(
stage="stream/normalized", raw_text="<empty-after-normalize>"
)
continue
_log_video_stream_line(stage="stream/normalized", raw_text=line)
try:
data = orjson.loads(line)
except orjson.JSONDecodeError:
_log_video_stream_line(stage="stream/non-json", raw_text=line)
continue
_log_raw_video_stream_event(line)
resp = data.get("result", {}).get("response", {})
is_thinking = bool(resp.get("isThinking"))
@@ -1438,6 +1620,17 @@ class VideoStreamProcessor(BaseProcessor):
if video_resp := resp.get("streamingVideoGenerationResponse"):
progress = video_resp.get("progress", 0)
latest_progress = max(latest_progress, int(progress or 0))
latest_video_post_id = str(
video_resp.get("videoPostId")
or video_resp.get("assetId")
or video_resp.get("videoId")
or latest_video_post_id
).strip()
latest_thumbnail_url = str(
video_resp.get("thumbnailImageUrl")
or latest_thumbnail_url
).strip()
if is_thinking:
if not self.show_think:
@@ -1481,6 +1674,7 @@ class VideoStreamProcessor(BaseProcessor):
f"post_id={video_post_id or '-'}, duration_ms={render_duration_ms:.2f}"
)
yield self._sse(rendered)
final_rendered = True
logger.info(f"Video generated: {video_url} (post_id={video_post_id})")
if video_post_id and self.token:
@@ -1492,15 +1686,73 @@ class VideoStreamProcessor(BaseProcessor):
)
continue
if (not final_rendered) and latest_progress >= 90 and latest_video_post_id:
await self._wait_before_eof_fallback(
post_id=latest_video_post_id,
progress=latest_progress,
wait_seconds=60,
)
rendered = await self._try_render_from_post_with_retry(
post_id=latest_video_post_id,
progress=latest_progress,
thumbnail_url=latest_thumbnail_url,
)
if rendered:
logger.info(
"Video generated via high-progress fallback: "
f"post_id={latest_video_post_id}, progress={latest_progress}"
)
if latest_video_post_id and self.token:
await _try_log_video_share_link(
self.token,
latest_video_post_id,
local_url=rendered,
thumbnail_url=latest_thumbnail_url,
)
yield self._sse(rendered)
final_rendered = True
_log_video_stream_end(stage="stream", reason="upstream_eof")
if self.think_opened:
yield self._sse("</think>\n")
yield self._sse(finish="stop")
yield "data: [DONE]\n\n"
except asyncio.CancelledError:
_log_video_stream_end(stage="stream", reason="cancelled")
logger.debug(
"Video stream cancelled by client", extra={"model": self.model}
)
except StreamIdleTimeoutError as e:
if (not final_rendered) and latest_progress >= 90 and latest_video_post_id:
rendered = await self._try_render_from_post_with_retry(
post_id=latest_video_post_id,
progress=latest_progress,
thumbnail_url=latest_thumbnail_url,
)
if rendered:
if self.think_opened:
yield self._sse("</think>\n")
self.think_opened = False
logger.info(
"Video generated via high-progress timeout fallback: "
f"post_id={latest_video_post_id}, progress={latest_progress}"
)
if latest_video_post_id and self.token:
await _try_log_video_share_link(
self.token,
latest_video_post_id,
local_url=rendered,
thumbnail_url=latest_thumbnail_url,
)
yield self._sse(rendered)
yield self._sse(finish="stop")
yield "data: [DONE]\n\n"
return
_log_video_stream_end(
stage="stream",
reason="idle_timeout",
extra=f"timeout={idle_timeout}",
)
raise AppException(
message="视频生成失败:网络连接异常,请稍后重试",
error_type=ErrorType.SERVER.value,
@@ -1508,6 +1760,11 @@ class VideoStreamProcessor(BaseProcessor):
status_code=504,
)
except RequestsError as e:
_log_video_stream_end(
stage="stream",
reason="request_error",
extra=f"error={e}",
)
if _is_http2_error(e):
logger.warning(
f"HTTP/2 stream error in video: {e}", extra={"model": self.model}
@@ -1528,6 +1785,11 @@ class VideoStreamProcessor(BaseProcessor):
status_code=502,
)
except Exception as e:
_log_video_stream_end(
stage="stream",
reason="exception",
extra=f"type={type(e).__name__}, error={e}",
)
logger.error(
f"Video stream processing error: {e}",
extra={"model": self.model, "error_type": type(e).__name__},
@@ -1671,13 +1933,25 @@ class VideoCollectProcessor(BaseProcessor):
try:
async for line in _with_idle_timeout(response, idle_timeout, self.model):
raw_line = (
line.decode("utf-8", errors="ignore")
if isinstance(line, (bytes, bytearray))
else str(line)
)
_log_video_stream_line(stage="collect/raw", raw_text=raw_line)
line = _normalize_line(line)
if not line:
_log_video_stream_line(
stage="collect/normalized", raw_text="<empty-after-normalize>"
)
continue
_log_video_stream_line(stage="collect/normalized", raw_text=line)
try:
data = orjson.loads(line)
except orjson.JSONDecodeError:
_log_video_stream_line(stage="collect/non-json", raw_text=line)
continue
_log_raw_video_stream_event(line)
resp = data.get("result", {}).get("response", {})
@@ -1735,15 +2009,31 @@ class VideoCollectProcessor(BaseProcessor):
fallback_video_id = fid
break
_log_video_stream_end(
stage="collect",
reason="upstream_eof",
extra=f"has_content={bool(content)}, fallback_video_id={fallback_video_id or '-'}",
)
except asyncio.CancelledError:
_log_video_stream_end(stage="collect", reason="cancelled")
logger.debug(
"Video collect cancelled by client", extra={"model": self.model}
)
except StreamIdleTimeoutError as e:
_log_video_stream_end(
stage="collect",
reason="idle_timeout",
extra=f"timeout={idle_timeout}, fallback_video_id={fallback_video_id or '-'}",
)
logger.warning(
f"Video collect idle timeout: {e}", extra={"model": self.model}
)
except RequestsError as e:
_log_video_stream_end(
stage="collect",
reason="request_error",
extra=f"error={e}, fallback_video_id={fallback_video_id or '-'}",
)
if _is_http2_error(e):
logger.warning(
f"HTTP/2 stream error in video collect: {e}",
@@ -1766,11 +2056,21 @@ class VideoCollectProcessor(BaseProcessor):
extra={"model": self.model},
)
raise
_log_video_stream_end(
stage="collect",
reason="upstream_exception",
extra=f"error={e}, fallback_video_id={fallback_video_id or '-'}",
)
logger.error(
f"Video collect upstream error: {e}",
extra={"model": self.model, "error_type": type(e).__name__},
)
except Exception as e:
_log_video_stream_end(
stage="collect",
reason="exception",
extra=f"type={type(e).__name__}, error={e}, fallback_video_id={fallback_video_id or '-'}",
)
logger.error(
f"Video collect processing error: {e}",
extra={"model": self.model, "error_type": type(e).__name__},
+59
View File
@@ -2,6 +2,8 @@
Local cache utilities.
"""
import json
import re
from typing import Any, Dict
from app.core.storage import DATA_DIR
@@ -26,6 +28,47 @@ class CacheService:
def _allowed_exts(self, media_type: str):
return IMAGE_EXTS if media_type == "image" else VIDEO_EXTS
def _media_meta_dir(self):
return DATA_DIR / "tmp" / "media-meta"
def _load_video_metadata(self) -> dict[str, dict[str, Any]]:
meta_dir = self._media_meta_dir()
if not meta_dir.exists():
return {}
metadata_by_post_id: dict[str, dict[str, Any]] = {}
for file_path in meta_dir.glob("*.json"):
try:
payload = json.loads(file_path.read_text(encoding="utf-8"))
except Exception:
continue
if not isinstance(payload, dict):
continue
if str(payload.get("media_type") or "").strip().lower() != "video":
continue
post_id = str(payload.get("post_id") or "").strip()
if not post_id:
continue
metadata_by_post_id[post_id] = payload
return metadata_by_post_id
@staticmethod
def _extract_post_id_from_name(name: str) -> str:
text = str(name or "").strip()
if not text:
return ""
match = re.search(
r"generated-([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})-",
text,
)
if match:
return match.group(1)
all_matches = re.findall(
r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}",
text,
)
return all_matches[-1] if all_matches else ""
def get_stats(self, media_type: str = "image") -> Dict[str, Any]:
cache_dir = self._cache_dir(media_type)
if not cache_dir.exists():
@@ -70,8 +113,24 @@ class CacheService:
start = max(0, (page - 1) * page_size)
paged = items[start : start + page_size]
metadata_by_post_id = self._load_video_metadata() if media_type == "video" else {}
for item in paged:
item["view_url"] = f"/v1/files/{media_type}/{item['name']}"
if media_type != "video":
continue
post_id = self._extract_post_id_from_name(item["name"])
if not post_id:
continue
metadata = metadata_by_post_id.get(post_id)
if not metadata:
continue
item["post_id"] = str(metadata.get("post_id") or "").strip()
item["share_link"] = str(metadata.get("share_link") or "").strip()
item["original_post_id"] = str(metadata.get("original_post_id") or "").strip()
item["media_url"] = str(metadata.get("media_url") or "").strip()
item["thumbnail_url"] = str(metadata.get("thumbnail_url") or "").strip()
item["local_url"] = str(metadata.get("local_url") or "").strip()
return {"total": total, "page": page, "page_size": page_size, "items": paged}
+132 -1
View File
@@ -9,6 +9,7 @@ import base64
import hashlib
import os
import time
import urllib.request
from pathlib import Path
from typing import List, Optional, Tuple
from urllib.parse import urlparse
@@ -21,6 +22,7 @@ from app.core.storage import DATA_DIR
from app.core.config import get_config
from app.core.exceptions import AppException
from app.services.reverse.assets_download import AssetsDownloadReverse
from app.services.reverse.utils.headers import build_headers
from app.services.grok.utils.locks import _get_download_semaphore, _file_lock
@@ -51,6 +53,14 @@ class DownloadService:
async def resolve_url(
self, path_or_url: str, token: str, media_type: str = "image"
) -> str:
if self._is_public_share_url(path_or_url):
app_url = get_config("app.app_url")
filename = self._public_cache_filename(path_or_url, media_type)
if app_url:
await self.download_file(path_or_url, token, media_type)
return f"{app_url.rstrip('/')}/v1/files/{media_type}/{filename}"
return path_or_url
asset_url = path_or_url
path = path_or_url
if path_or_url.startswith("http"):
@@ -69,6 +79,120 @@ class DownloadService:
return f"{app_url.rstrip('/')}/v1/files/{media_type}{path}"
return asset_url
@staticmethod
def _is_public_share_url(url: str) -> bool:
text = str(url or "").strip().lower()
return (
text.startswith("https://imagine-public.x.ai/")
or text.startswith("http://imagine-public.x.ai/")
)
@staticmethod
def _is_localhost_url(url: str) -> bool:
parsed = urlparse(str(url or "").strip())
host = (parsed.hostname or "").lower()
return host in {"localhost", "127.0.0.1", "::1"}
@staticmethod
def _public_cache_filename(file_url: str, media_type: str = "image") -> str:
parsed = urlparse(str(file_url or "").strip())
suffix = Path(parsed.path or "").suffix
filename = (
(parsed.netloc + (parsed.path or ""))
.lstrip("/")
.replace("/", "-")
.replace(":", "-")
)
if not filename:
filename = f"public-{media_type}"
if suffix and not filename.endswith(suffix):
filename = f"{filename}{suffix}"
return filename
async def _download_public_url(
self,
file_url: str,
media_type: str = "image",
) -> Tuple[Optional[Path], str]:
"""下载公开分享直链,避免误走 assets.grok.com 反代。"""
started_at = time.perf_counter()
cache_dir = self.image_dir if media_type == "image" else self.video_dir
filename = self._public_cache_filename(file_url, media_type)
cache_path = cache_dir / filename
logger.info(
f"Public download started: media_type={media_type}, file_url={file_url}, cache_path={cache_path}"
)
lock_name = (
f"dl_public_{media_type}_{hashlib.sha1(str(cache_path).encode()).hexdigest()[:16]}"
)
lock_timeout = max(1, int(get_config("asset.download_timeout")))
async with _file_lock(lock_name, timeout=lock_timeout):
base_proxy = (get_config("proxy.base_proxy_url") or "").strip()
asset_proxy = (get_config("proxy.asset_proxy_url") or "").strip()
proxy_url = asset_proxy or base_proxy
headers = build_headers(
cookie_token="",
content_type="video/mp4" if media_type == "video" else "image/jpeg",
origin="https://grok.com",
referer="https://grok.com/",
)
if media_type == "video":
headers["Range"] = "bytes=0-"
def _urllib_download():
opener = (
urllib.request.build_opener(
urllib.request.ProxyHandler(
{"http": proxy_url, "https": proxy_url}
)
)
if proxy_url
else urllib.request.build_opener()
)
req = urllib.request.Request(file_url, headers=headers, method="GET")
with opener.open(req, timeout=get_config("asset.download_timeout")) as resp:
status = int(getattr(resp, "status", 200) or 200)
body = resp.read()
raw_headers = {str(k).lower(): str(v) for k, v in dict(resp.headers.items()).items()}
return status, raw_headers, body
status_code, response_headers, response_body = await asyncio.to_thread(
_urllib_download
)
if status_code not in (200, 206):
raise AppException(
f"Public download failed, status={status_code}",
code="public_download_failed",
)
logger.info(
f"Public download response received: media_type={media_type}, file_url={file_url}, "
f"status={status_code}, content_type={(response_headers.get('content-type', 'application/octet-stream').split(';')[0])}"
)
tmp_path = cache_path.with_suffix(cache_path.suffix + ".tmp")
try:
async with aiofiles.open(tmp_path, "wb") as f:
await f.write(response_body)
os.replace(tmp_path, cache_path)
finally:
if tmp_path.exists() and not cache_path.exists():
try:
tmp_path.unlink()
except Exception:
pass
mime = response_headers.get(
"content-type", "application/octet-stream"
).split(";")[0]
duration_ms = (time.perf_counter() - started_at) * 1000
logger.info(
f"Public downloaded: {file_url}, media_type={media_type}, cache_path={cache_path}, "
f"mime={mime}, duration_ms={duration_ms:.2f}"
)
asyncio.create_task(self._check_limit())
return cache_path, mime
async def render_image(
self, url: str, token: str, image_id: str = "image"
) -> str:
@@ -101,7 +225,11 @@ class DownloadService:
final_video_url = await self.resolve_url(video_url, token, "video")
final_thumb_url = ""
if thumbnail_url:
final_thumb_url = await self.resolve_url(thumbnail_url, token, "image")
try:
final_thumb_url = await self.resolve_url(thumbnail_url, token, "image")
except Exception as e:
logger.warning(f"Render video thumbnail resolve failed, ignore thumbnail: {e}")
final_thumb_url = ""
duration_ms = (time.perf_counter() - started_at) * 1000
logger.info(
f"Render video resolved: final_video_url={final_video_url}, "
@@ -196,6 +324,9 @@ class DownloadService:
Returns:
Tuple[Optional[Path], str]: The path of the downloaded file and the MIME type.
"""
if self._is_public_share_url(file_path):
return await self._download_public_url(file_path, media_type)
started_at = time.perf_counter()
async with _get_download_semaphore():
file_path = self._normalize_path(file_path)
+19 -4
View File
@@ -44,12 +44,26 @@ class AssetsDownloadReverse:
yield data[i:i + chunk_size]
@staticmethod
async def _urllib_get(url: str, headers: dict[str, str], timeout: int) -> "AssetsDownloadReverse._SimpleResponse":
async def _urllib_get(
url: str,
headers: dict[str, str],
timeout: int,
proxy_url: str = "",
) -> "AssetsDownloadReverse._SimpleResponse":
"""使用标准库 urllib 兜底下载,绕过 curl TLS 实现。"""
req = urllib.request.Request(url=url, headers=headers, method="GET")
def _do_get():
with urllib.request.urlopen(req, timeout=timeout) as resp:
opener = None
if proxy_url:
proxy_handler = urllib.request.ProxyHandler(
{"http": proxy_url, "https": proxy_url}
)
opener = urllib.request.build_opener(proxy_handler)
else:
opener = urllib.request.build_opener()
with opener.open(req, timeout=timeout) as resp:
status = int(getattr(resp, "status", 200) or 200)
body = resp.read()
raw_headers = {}
@@ -137,13 +151,13 @@ class AssetsDownloadReverse:
try:
return await _single_get(use_impersonate=True, use_proxy=True)
except Exception as first_err:
# TLS/握手类问题常出现在代理或浏览器指纹组合,降级直连重试一次。
# TLS/握手类问题常出现在浏览器指纹组合,降级为不伪装浏览器后重试。
logger.warning(
"AssetsDownloadReverse primary request failed, fallback direct: "
f"error={first_err}"
)
try:
return await _single_get(use_impersonate=False, use_proxy=False)
return await _single_get(use_impersonate=False, use_proxy=True)
except Exception as second_err:
# curl 栈仍失败时,使用标准库 urllib 再兜底一次,绕开 curl TLS 实现。
logger.warning(
@@ -154,6 +168,7 @@ class AssetsDownloadReverse:
url=url,
headers=headers,
timeout=timeout,
proxy_url=proxy_url,
)
if fallback_resp.status_code != 200:
logger.error(
+9 -2
View File
@@ -4,6 +4,7 @@ Reverse interface: media post create/get.
import asyncio
import json
import re
import urllib.request
from pathlib import Path
from dataclasses import dataclass
@@ -87,9 +88,15 @@ class MediaPostReverse:
except Exception as e:
logger.warning(f"MediaPost metadata create-link failed: post_id={post_text}, error={e}")
canonical_post_id = post_text
if share_link:
match = re.search(r"/imagine/post/([0-9a-fA-F-]{32,36})", share_link)
if match:
canonical_post_id = match.group(1)
post = post_payload.get("post", {}) if isinstance(post_payload, dict) else {}
metadata: dict[str, Any] = {
"post_id": post_text,
"post_id": canonical_post_id,
"share_link": share_link,
"media_type": str(media_type or post.get("mediaType") or "").strip(),
"media_url": str(post.get("mediaUrl") or "").strip(),
@@ -103,7 +110,7 @@ class MediaPostReverse:
}
if extra:
metadata.update(extra)
path = await MediaPostReverse.write_metadata(post_text, metadata)
path = await MediaPostReverse.write_metadata(canonical_post_id, metadata)
if path:
metadata["metadata_path"] = str(path)
return metadata
+83 -35
View File
@@ -209,19 +209,59 @@
if (editExtendPostId) editExtendPostId.textContent = shortHash(currentExtendPostId);
}
const UUID_RE = /[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}/g;
// 从缓存视频文件名中提取 parentPostId
// 文件名格式示例: users-xxx-generated-{postId}-generated_video_hd.mp4
function extractPostIdFromFileName(name) {
const s = String(name || '').trim();
if (!s) return '';
// 尝试 generated-{uuid}- 模式
const m = s.match(/generated-([0-9a-fA-F-]{32,36})-/);
const m = s.match(/generated-([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})-/);
if (m) return m[1];
// 回退:匹配最后一个 UUID 格式
const allUuids = s.match(/[0-9a-fA-F-]{32,36}/g);
const allUuids = s.match(UUID_RE);
return allUuids && allUuids.length ? allUuids[allUuids.length - 1] : '';
}
function extractPostIdFromShareLink(link) {
const text = String(link || '').trim();
if (!text) return '';
const match = text.match(/\/imagine\/post\/([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})(?:[/?#]|$)/i);
return match ? match[1] : '';
}
function resolveVideoPostId(meta = {}) {
const directPostId = String(meta.postId || meta.post_id || '').trim();
if (directPostId) return directPostId;
const shareLinkPostId = extractPostIdFromShareLink(meta.shareLink || meta.share_link || '');
if (shareLinkPostId) return shareLinkPostId;
const originalPostId = String(meta.originalPostId || meta.original_post_id || '').trim();
if (originalPostId) return originalPostId;
const namePostId = extractPostIdFromFileName(meta.name || '');
if (namePostId) return namePostId;
return extractPostIdFromFileName(meta.url || '');
}
function applyResolvedVideoIdentity(meta = {}, sourceLabel = '') {
const resolvedPostId = resolveVideoPostId(meta);
if (!resolvedPostId) {
if (sourceLabel) {
console.warn(`[视频标识] ${sourceLabel} 未解析到 post_id`, meta);
}
return '';
}
currentExtendPostId = resolvedPostId;
currentFileAttachmentId = resolvedPostId;
if (!originalFileAttachmentId) {
originalFileAttachmentId = resolvedPostId;
}
if (sourceLabel) {
console.log(`[视频标识] ${sourceLabel} 解析 post_id:`, resolvedPostId);
}
return resolvedPostId;
}
function debugLog(...args) {
// console.log('[video-extend-debug]', ...args);
}
@@ -1570,24 +1610,13 @@
});
}
function bindEditVideoSource(url) {
function bindEditVideoSource(url, meta = {}) {
const safeUrl = String(url || '').trim();
selectedVideoUrl = safeUrl;
if (editHint) {
editHint.classList.toggle('hidden', Boolean(safeUrl));
}
// 从 URL 中提取 postId (支持历史面板点击编辑)
const postId = extractPostIdFromFileName(safeUrl);
if (postId) {
currentExtendPostId = postId;
currentFileAttachmentId = postId;
// 首次设置 originalFileAttachmentId(后续延长不覆盖)
if (!originalFileAttachmentId) {
originalFileAttachmentId = postId;
debugLog('bindEditVideoSource: set originalFileAttachmentId =', postId);
}
debugLog('bindEditVideoSource: extracted postId =', postId);
}
applyResolvedVideoIdentity({ ...meta, url: safeUrl }, 'bindEditVideoSource');
if (!editVideo) return;
enforceInlinePlayback(editVideo);
editVideo.src = safeUrl;
@@ -1617,7 +1646,12 @@
}
if (editHint) editHint.classList.add('hidden');
if (editBody) editBody.classList.remove('hidden');
bindEditVideoSource(url);
bindEditVideoSource(url, item ? {
postId: item.dataset.postId || '',
shareLink: item.dataset.shareLink || '',
originalPostId: item.dataset.originalPostId || '',
name: item.dataset.name || '',
} : {});
}
function closeEditPanel() {
@@ -1747,9 +1781,12 @@
const html = items.map((item, idx) => {
const name = String(item.name || '');
const url = String(item.view_url || '');
const postId = String(item.post_id || '');
const shareLink = String(item.share_link || '');
const originalPostId = String(item.original_post_id || '');
const size = formatBytes(item.size_bytes);
const mtime = formatMtime(item.mtime_ms);
return `<div class="cache-video-item" data-url="${url}" data-name="${name}">
return `<div class="cache-video-item" data-url="${url}" data-name="${name}" data-post-id="${postId}" data-share-link="${shareLink}" data-original-post-id="${originalPostId}">
<div class="cache-video-thumb-wrap">
<video class="cache-video-thumb" src="${url}" preload="auto" muted playsinline></video>
</div>
@@ -1788,25 +1825,20 @@
}
}
function useCachedVideo(url, name) {
function useCachedVideo(url, name, meta = {}) {
const safeUrl = String(url || '').trim();
if (!safeUrl) return;
selectedVideoItemId = `cache-${Date.now()}`;
selectedVideoUrl = safeUrl;
// 从文件名中自动提取 parentPostId 用于视频延长
const extractedPostId = extractPostIdFromFileName(String(name || ''));
if (extractedPostId) {
currentExtendPostId = extractedPostId;
currentFileAttachmentId = extractedPostId;
// 从缓存重新选择视频时重置 originalFileAttachmentId(开启新的延长链)
originalFileAttachmentId = extractedPostId;
debugLog('useCachedVideo: extracted postId =', extractedPostId);
}
originalFileAttachmentId = '';
applyResolvedVideoIdentity({ ...meta, name, url: safeUrl }, 'useCachedVideo');
if (imageUrlInput) imageUrlInput.value = safeUrl;
if (imageFileName && name) imageFileName.textContent = name;
if (enterEditBtn) enterEditBtn.disabled = false;
closeCacheVideoModal();
openEditPanel();
bindEditVideoSource(safeUrl, { ...meta, name });
if (editHint) editHint.classList.add('hidden');
if (editBody) editBody.classList.remove('hidden');
setEditMeta();
}
@@ -2404,11 +2436,12 @@
editVideo.src = videoUrl;
editVideo.load();
}
// 从新视频 URL 提取 postId 用于链式延长
const newPostId = extractPostIdFromFileName(videoUrl);
// 优先从 create-link 元数据对应的 post_id,最后才回退到 URL
const newPostId = applyResolvedVideoIdentity({ url: videoUrl }, 'extend-sse');
if (newPostId) {
currentExtendPostId = newPostId; // 更新当前 postId
currentFileAttachmentId = newPostId;
if (item) {
item.dataset.postId = newPostId;
}
console.log('[SSE 调试] 从新视频成功提取到新的 extend_post_id:', newPostId);
} else {
console.warn('[SSE 调试] 未能从新视频地址提取出新的 extend_post_id!', videoUrl);
@@ -2708,7 +2741,11 @@
if (!target.classList.contains('cache-video-use')) return;
const row = target.closest('.cache-video-item');
if (!row) return;
useCachedVideo(row.getAttribute('data-url') || '', row.getAttribute('data-name') || '');
useCachedVideo(row.getAttribute('data-url') || '', row.getAttribute('data-name') || '', {
postId: row.getAttribute('data-post-id') || '',
shareLink: row.getAttribute('data-share-link') || '',
originalPostId: row.getAttribute('data-original-post-id') || '',
});
});
}
@@ -2726,7 +2763,13 @@
return;
}
// 提取 postId 用于延长
const postId = extractPostIdFromFileName(bUrl);
const postId = resolveVideoPostId({
postId: item.dataset.postId || '',
shareLink: item.dataset.shareLink || '',
originalPostId: item.dataset.originalPostId || '',
name: item.dataset.name || '',
url: bUrl,
});
if (postId) {
currentExtendPostId = postId;
currentFileAttachmentId = postId;
@@ -2750,7 +2793,12 @@
return;
}
if (!target.classList.contains('video-download')) {
bindEditVideoSource(selectedVideoUrl);
bindEditVideoSource(selectedVideoUrl, {
postId: item.dataset.postId || '',
shareLink: item.dataset.shareLink || '',
originalPostId: item.dataset.originalPostId || '',
name: item.dataset.name || '',
});
return;
}
event.preventDefault();
-2
View File
@@ -124,8 +124,6 @@ concurrent = 100
timeout = 60
# 流式空闲超时时间(秒)
stream_timeout = 60
# 视频延长流式空闲超时时间(秒)
extension_stream_timeout = 300
# ==================== 语音配置 ====================
[voice]