a3f61fcc0f
- DB: segments 用户输入文本（Alembic 0002） - Chat: 阶段检测/阶段提示/回复限制，编排与访谈/画像 prompts 调整 - Memoir: 忠实度检查 agent，叙事与分类等链路更新 - Core: agent 日志、Alembic 启动、LangChain/日志/配置等 - Story: time_hints；Memory 检索与相关测试 - Expo: 助手头像、会话页与消息拆分、实时会话与文案/i18n - Docs/scripts/tests: 迁移脚本、LLM JSON/记忆检索文档、新增单测
45 lines
1.4 KiB
Python
45 lines
1.4 KiB
Python
"""按 Asset id 批量生成 COS 签名 URL（解析正文 asset://）。"""
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.config import settings
|
||
from app.core.logging import get_logger
|
||
from app.features.asset.models import Asset
|
||
from app.features.memoir.memoir_images.storage import (
|
||
CosDownloadUrlError,
|
||
TencentCosStorageService,
|
||
)
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
async def signed_urls_for_asset_ids(
|
||
db: AsyncSession, asset_ids: set[str]
|
||
) -> dict[str, str]:
|
||
"""返回 asset_id -> 短期可访问 URL；签名失败则跳过该 id。"""
|
||
if not asset_ids:
|
||
return {}
|
||
stmt = select(Asset).where(Asset.id.in_(asset_ids))
|
||
result = await db.execute(stmt)
|
||
rows = list(result.scalars().all())
|
||
storage = TencentCosStorageService.from_settings(settings)
|
||
out: dict[str, str] = {}
|
||
for a in rows:
|
||
key = (a.storage_key or "").strip()
|
||
if not key:
|
||
continue
|
||
try:
|
||
out[a.id] = storage.get_download_url(key)
|
||
except CosDownloadUrlError as exc:
|
||
logger.warning(
|
||
"Asset 签名失败: id={} key={} retryable={} error={}",
|
||
a.id,
|
||
key,
|
||
exc.retryable,
|
||
exc,
|
||
)
|
||
except Exception as exc:
|
||
logger.warning("Asset 签名失败: id={} error={}", a.id, exc)
|
||
return out
|