fix(chat): 链接卡片补全公众号来源并解决缩略图防盗链

- appmsg 解析补全 from/fromUsername,并规范化 url/thumbUrl
- contact.db 兜底反查 fromUsername(仅有 sourcedisplayname 时)
- 新增 /api/chat/media/proxy_image,仅允许 qpic/qlogo,带 mp.weixin.qq.com Referer(10MB 限制)
- 前端 LinkCard 增加来源头像/host 兜底,qpic/qlogo 预览图走代理;头像加载失败回退
- 导出消息补充 from 字段
This commit is contained in:
2977094657
2026-01-24 10:51:35 +08:00
parent d3d1c8dc7d
commit c523036a10
5 changed files with 510 additions and 34 deletions

View File

@@ -289,7 +289,13 @@
<!-- 消息发送者头像 -->
<div class="w-[36px] h-[36px] rounded-md overflow-hidden bg-gray-300 flex-shrink-0" :class="[message.isSent ? 'ml-3' : 'mr-3', { 'privacy-blur': privacyMode }]">
<div v-if="message.avatar" class="w-full h-full">
<img :src="message.avatar" :alt="message.sender + '的头像'" class="w-full h-full object-cover">
<img
:src="message.avatar"
:alt="message.sender + '的头像'"
class="w-full h-full object-cover"
referrerpolicy="no-referrer"
@error="onMessageAvatarError($event, message)"
>
</div>
<div v-else class="w-full h-full flex items-center justify-center text-white text-xs font-bold"
:style="{ backgroundColor: message.avatarColor || (message.isSent ? '#4B5563' : '#6B7280') }">
@@ -319,7 +325,9 @@
:heading="message.title || message.content"
:abstract="message.content"
:preview="message.preview"
:fromAvatar="message.fromAvatar"
:from="message.from"
:isSent="message.isSent"
/>
<div v-else-if="message.renderType === 'file'"
class="wechat-redpacket-card wechat-special-card wechat-file-card msg-radius"
@@ -3912,6 +3920,36 @@ const normalizeMessage = (msg) => {
)
}
// WeChat public account thumbnails (mmbiz.qpic.cn, wx.qlogo.cn...) are hotlink-protected:
// the browser will get a placeholder image ("此图片来自微信公众号平台").
// Proxy them via backend with a mp.weixin.qq.com Referer to fetch the real image.
const normalizedThumbUrl = (() => {
// Backend may provide either `thumbUrl` (appmsg) or `preview` (some exports). Use the first usable one.
const candidates = [msg.thumbUrl, msg.preview]
for (const cand of candidates) {
if (isUsableMediaUrl(cand)) return normalizeMaybeUrl(cand)
}
return ''
})()
const normalizedLinkPreviewUrl = (() => {
const u = normalizedThumbUrl
if (!u) return ''
if (/^\/api\/chat\/media\//i.test(u) || /^blob:/i.test(u) || /^data:/i.test(u)) return u
if (!/^https?:\/\//i.test(u)) return u
try {
const host = new URL(u).hostname.toLowerCase()
if (host.endsWith('.qpic.cn') || host.endsWith('.qlogo.cn')) {
return `${mediaBase}/api/chat/media/proxy_image?url=${encodeURIComponent(u)}`
}
} catch {}
return u
})()
const fromUsername = String(msg.fromUsername || '').trim()
const fromAvatar = fromUsername
? `${mediaBase}/api/chat/avatar?account=${encodeURIComponent(selectedAccount.value || '')}&username=${encodeURIComponent(fromUsername)}`
: ''
const localEmojiUrl = msg.emojiMd5 ? `${mediaBase}/api/chat/media/emoji?account=${encodeURIComponent(selectedAccount.value || '')}&md5=${encodeURIComponent(msg.emojiMd5)}&username=${encodeURIComponent(selectedContact.value?.username || '')}` : ''
const localImageUrl = (() => {
if (!msg.imageMd5 && !msg.imageFileId) return ''
@@ -4051,14 +4089,23 @@ const normalizeMessage = (msg) => {
transferReceived: msg.paySubType === '3' || msg.transferStatus === '已收款',
voiceUrl: normalizedVoiceUrl || '',
voiceDuration: msg.voiceLength || msg.voiceDuration || '',
preview: msg.thumbUrl || '',
from: '',
preview: normalizedLinkPreviewUrl || '',
from: String(msg.from || '').trim(),
fromUsername,
fromAvatar,
isGroup: !!selectedContact.value?.isGroup,
avatar: msg.senderAvatar || fallbackAvatar || null,
// Backends may use either `senderAvatar` (our API) or `avatar` (exported JSON).
avatar: msg.senderAvatar || msg.avatar || fallbackAvatar || null,
avatarColor: null
}
}
const onMessageAvatarError = (e, message) => {
// Make sure we fall back to the initial avatar if the URL 404s/blocks.
try { e?.target && (e.target.style.display = 'none') } catch {}
try { if (message) message.avatar = null } catch {}
}
const shouldShowEmojiDownload = (message) => {
if (!message?.emojiMd5) return false
const u = String(message?.emojiRemoteUrl || '').trim()
@@ -4989,29 +5036,90 @@ const LinkCard = defineComponent({
heading: { type: String, default: '' },
abstract: { type: String, default: '' },
preview: { type: String, default: '' },
from: { type: String, default: '' }
fromAvatar: { type: String, default: '' },
from: { type: String, default: '' },
isSent: { type: Boolean, default: false }
},
setup(props) {
return () => h(
const getFromText = () => {
const raw = String(props.from || '').trim()
if (raw) return raw
// Fallback: when the appmsg XML doesn't provide sourcedisplayname/appname,
// show the host so the footer row still matches WeChat's fixed card layout.
try {
const host = new URL(String(props.href || '')).hostname
return String(host || '').trim()
} catch {
return ''
}
}
return () => {
const fromText = getFromText()
// WeChat link cards show a small avatar next to the source text. We don't
// always have a real image URL, so fall back to the first glyph.
const fromAvatarText = (() => {
const t = String(fromText || '').trim()
return t ? (Array.from(t)[0] || '') : ''
})()
const fromAvatarUrl = String(props.fromAvatar || '').trim()
return h(
'a',
{
href: props.href,
target: '_blank',
rel: 'noreferrer',
class: 'block max-w-sm w-full bg-white msg-radius border border-neutral-200 overflow-hidden hover:bg-gray-50 transition-colors'
class: [
'wechat-link-card',
'wechat-special-card',
'msg-radius',
props.isSent ? 'wechat-special-sent-side' : ''
].filter(Boolean).join(' '),
// Inline size is intentional: LinkCard is a local component rendered via `h()` and
// does not inherit the SFC scoped CSS attribute, so relying on scoped CSS for exact
// sizing is fragile. Keep width in sync with the WeChat desktop card size.
style: {
width: '210px',
minWidth: '210px',
maxWidth: '210px',
display: 'flex',
flexDirection: 'column',
boxSizing: 'border-box',
flex: '0 0 auto',
background: '#fff',
border: 'none',
boxShadow: 'none',
textDecoration: 'none',
outline: 'none'
}
},
[
props.preview ? h('div', { class: 'w-full bg-black/5' }, [
h('img', { src: props.preview, alt: props.heading || '链接预览', class: 'w-full max-h-40 object-cover' })
]) : null,
h('div', { class: 'px-3 py-2' }, [
h('div', { class: 'text-sm font-medium text-gray-900 line-clamp-2' }, props.heading || props.href),
props.abstract ? h('div', { class: 'text-xs text-gray-600 mt-1 line-clamp-2' }, props.abstract) : null,
props.from ? h('div', { class: 'text-[10px] text-gray-400 mt-1 truncate' }, props.from) : null
h('div', { class: 'wechat-link-content' }, [
h('div', { class: 'wechat-link-info' }, [
h('div', { class: 'wechat-link-title' }, props.heading || props.href),
props.abstract ? h('div', { class: 'wechat-link-desc' }, props.abstract) : null
].filter(Boolean)),
props.preview ? h('div', { class: 'wechat-link-thumb' }, [
h('img', { src: props.preview, alt: props.heading || '链接预览', class: 'wechat-link-thumb-img', referrerpolicy: 'no-referrer' })
]) : null
].filter(Boolean)),
h('div', { class: 'wechat-link-from' }, [
h('div', { class: 'wechat-link-from-avatar', 'aria-hidden': 'true' }, [
fromAvatarText || '\u200B',
fromAvatarUrl ? h('img', {
src: fromAvatarUrl,
alt: '',
class: 'wechat-link-from-avatar-img',
referrerpolicy: 'no-referrer',
onError: (e) => { try { e?.target && (e.target.style.display = 'none') } catch {} }
}) : null
].filter(Boolean)),
h('div', { class: 'wechat-link-from-name' }, fromText || '\u200B')
])
].filter(Boolean)
)
}
}
})
</script>
@@ -5324,24 +5432,24 @@ const LinkCard = defineComponent({
}
/* 统一特殊消息尾巴(红包 / 文件等) */
.wechat-special-card {
:deep(.wechat-special-card) {
position: relative;
overflow: visible;
}
.wechat-special-card::after {
:deep(.wechat-special-card)::after {
content: '';
position: absolute;
top: 16px;
top: 12px;
left: -4px;
width: 10px;
height: 10px;
width: 12px;
height: 12px;
background-color: inherit;
transform: rotate(45deg);
border-radius: 2px;
}
.wechat-special-sent-side::after {
:deep(.wechat-special-sent-side)::after {
left: auto;
right: -4px;
}
@@ -5693,6 +5801,138 @@ const LinkCard = defineComponent({
margin-right: 4px;
}
/* 链接消息样式 - 微信风格 */
:deep(.wechat-link-card) {
width: 210px;
min-width: 210px;
max-width: 210px;
background: #fff;
display: flex;
flex-direction: column;
box-sizing: border-box;
border: none;
box-shadow: none;
outline: none;
cursor: pointer;
text-decoration: none;
transition: background-color 0.15s ease;
}
:deep(.wechat-link-card:hover) {
background: #f5f5f5;
}
:deep(.wechat-link-content) {
display: flex;
flex-direction: row;
align-items: flex-start;
gap: 10px;
box-sizing: border-box;
/* Keep a small breathing room above the footer divider. */
padding: 8px 10px 6px;
flex: 1 1 auto;
}
:deep(.wechat-link-info) {
display: flex;
flex-direction: column;
overflow: hidden;
flex: 1 1 auto;
min-width: 0;
}
:deep(.wechat-link-title) {
font-size: 14px;
color: #1a1a1a;
display: -webkit-box;
-webkit-line-clamp: 2;
-webkit-box-orient: vertical;
overflow: hidden;
line-height: 1.4;
word-break: break-word;
}
:deep(.wechat-link-desc) {
font-size: 12px;
color: #8c8c8c;
margin-top: 4px;
display: -webkit-box;
-webkit-line-clamp: 3;
-webkit-box-orient: vertical;
overflow: hidden;
line-height: 1.4;
word-break: break-word;
}
:deep(.wechat-link-thumb) {
width: 42px;
height: 42px;
flex-shrink: 0;
border-radius: 0;
overflow: hidden;
background: #f2f2f2;
/* Center the thumbnail in the content area (WeChat desktop style). */
align-self: center;
}
:deep(.wechat-link-thumb-img) {
width: 100%;
height: 100%;
object-fit: cover;
display: block;
}
:deep(.wechat-link-from) {
height: 30px;
display: flex;
align-items: center;
gap: 5px;
padding: 0 10px;
position: relative;
flex-shrink: 0;
}
:deep(.wechat-link-from)::before {
content: '';
position: absolute;
top: 0;
left: 11px;
right: 11px;
height: 1.5px;
background: #e8e8e8;
}
:deep(.wechat-link-from-avatar) {
width: 16px;
height: 16px;
border-radius: 50%;
background: #111;
color: #fff;
font-size: 11px;
line-height: 16px;
text-align: center;
flex-shrink: 0;
position: relative;
overflow: hidden;
}
:deep(.wechat-link-from-avatar-img) {
position: absolute;
inset: 0;
width: 100%;
height: 100%;
object-fit: cover;
display: block;
}
:deep(.wechat-link-from-name) {
font-size: 12px;
color: #b2b2b2;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
/* 隐私模式模糊效果 */
.privacy-blur {
filter: blur(9px);

View File

@@ -894,6 +894,7 @@ def _parse_message_for_export(
content_text = raw_text
title = ""
url = ""
from_name = ""
record_item = ""
image_md5 = ""
image_file_id = ""
@@ -934,6 +935,7 @@ def _parse_message_for_export(
content_text = str(parsed.get("content") or "")
title = str(parsed.get("title") or "")
url = str(parsed.get("url") or "")
from_name = str(parsed.get("from") or "")
record_item = str(parsed.get("recordItem") or "")
quote_title = str(parsed.get("quoteTitle") or "")
quote_content = str(parsed.get("quoteContent") or "")
@@ -1162,6 +1164,7 @@ def _parse_message_for_export(
"content": content_text,
"title": title,
"url": url,
"from": from_name,
"recordItem": record_item,
"thumbUrl": thumb_url,
"imageMd5": image_md5,

View File

@@ -773,7 +773,21 @@ def _parse_app_message(text: str) -> dict[str, Any]:
app_type = 0
title = _extract_xml_tag_text(text, "title")
des = _extract_xml_tag_text(text, "des")
url = _extract_xml_tag_text(text, "url")
url = _normalize_xml_url(_extract_xml_tag_text(text, "url"))
# Some appmsg payloads (notably mp.weixin.qq.com link shares) include a "source" block:
# <sourceusername>gh_xxx</sourceusername>
# <sourcedisplayname>公众号名</sourcedisplayname>
# We'll surface that as `from` so the frontend can render the publisher line like WeChat.
source_display_name = (
_extract_xml_tag_text(text, "sourcedisplayname")
or _extract_xml_tag_text(text, "sourceDisplayName")
or _extract_xml_tag_text(text, "appname")
)
source_username = (
_extract_xml_tag_text(text, "sourceusername")
or _extract_xml_tag_text(text, "sourceUsername")
)
lower = text.lower()
@@ -794,13 +808,15 @@ def _parse_app_message(text: str) -> dict[str, Any]:
}
if app_type in (5, 68) and url:
thumb_url = _extract_xml_tag_text(text, "thumburl")
thumb_url = _normalize_xml_url(_extract_xml_tag_text(text, "thumburl"))
return {
"renderType": "link",
"content": des or title or "[链接]",
"title": title or des or "",
"url": url,
"thumbUrl": thumb_url or "",
"from": str(source_display_name or "").strip(),
"fromUsername": str(source_username or "").strip(),
}
if app_type in (6, 74):
@@ -1322,6 +1338,58 @@ def _load_contact_rows(contact_db_path: Path, usernames: list[str]) -> dict[str,
conn.close()
def _load_usernames_by_display_names(contact_db_path: Path, names: list[str]) -> dict[str, str]:
"""Best-effort mapping from display name -> username using contact.db.
Some appmsg/link payloads only provide `sourcedisplayname` (surfaced as `from`) but not
`sourceusername` (`fromUsername`). We use this mapping to recover `fromUsername` so the
frontend can render the publisher avatar via `/api/chat/avatar`.
"""
uniq = list(dict.fromkeys([str(n or "").strip() for n in names if str(n or "").strip()]))
if not uniq:
return {}
placeholders = ",".join(["?"] * len(uniq))
hits: dict[str, set[str]] = {}
conn = sqlite3.connect(str(contact_db_path))
conn.row_factory = sqlite3.Row
try:
def query_table(table: str) -> None:
for col in ("remark", "nick_name", "alias"):
sql = f"""
SELECT username, {col} AS display_name
FROM {table}
WHERE {col} IN ({placeholders})
"""
try:
rows = conn.execute(sql, uniq).fetchall()
except Exception:
rows = []
for r in rows:
try:
dn = str(r["display_name"] or "").strip()
u = str(r["username"] or "").strip()
except Exception:
continue
if not dn or not u:
continue
hits.setdefault(dn, set()).add(u)
query_table("contact")
query_table("stranger")
# Only return unambiguous mappings (display name -> exactly 1 username).
out: dict[str, str] = {}
for dn, users in hits.items():
if len(users) == 1:
out[dn] = next(iter(users))
return out
finally:
conn.close()
def _make_search_tokens(q: str) -> list[str]:
tokens = [t for t in re.split(r"\s+", str(q or "").strip()) if t]
if len(tokens) > 8:

View File

@@ -39,6 +39,7 @@ from ..chat_helpers import (
_make_snippet,
_match_tokens,
_load_contact_rows,
_load_usernames_by_display_names,
_load_latest_message_previews,
_lookup_resource_md5,
_normalize_xml_url,
@@ -1519,6 +1520,8 @@ def _append_full_messages_from_rows(
content_text = raw_text
title = ""
url = ""
from_name = ""
from_username = ""
record_item = ""
image_md5 = ""
emoji_md5 = ""
@@ -1561,6 +1564,8 @@ def _append_full_messages_from_rows(
content_text = str(parsed.get("content") or "")
title = str(parsed.get("title") or "")
url = str(parsed.get("url") or "")
from_name = str(parsed.get("from") or "")
from_username = str(parsed.get("fromUsername") or "")
record_item = str(parsed.get("recordItem") or "")
quote_title = str(parsed.get("quoteTitle") or "")
quote_content = str(parsed.get("quoteContent") or "")
@@ -1781,6 +1786,7 @@ def _append_full_messages_from_rows(
amount = str(parsed.get("amount") or amount)
cover_url = str(parsed.get("coverUrl") or cover_url)
thumb_url = str(parsed.get("thumbUrl") or thumb_url)
from_name = str(parsed.get("from") or from_name)
file_size = str(parsed.get("size") or file_size)
pay_sub_type = str(parsed.get("paySubType") or pay_sub_type)
file_md5 = str(parsed.get("fileMd5") or file_md5)
@@ -1828,6 +1834,8 @@ def _append_full_messages_from_rows(
"content": content_text,
"title": title,
"url": url,
"from": from_name,
"fromUsername": from_username,
"recordItem": record_item,
"imageMd5": image_md5,
"imageFileId": image_file_id,
@@ -1949,13 +1957,42 @@ def _postprocess_full_messages(
is_sent = m.get("isSent", False)
m["transferStatus"] = "已收款" if is_sent else "已被接收"
# Some appmsg payloads provide only `from` (sourcedisplayname) but not `fromUsername` (sourceusername).
# Recover `fromUsername` via contact.db so the frontend can render the publisher avatar.
missing_from_names = [
str(m.get("from") or "").strip()
for m in merged
if str(m.get("renderType") or "").strip() == "link"
and str(m.get("from") or "").strip()
and not str(m.get("fromUsername") or "").strip()
]
if missing_from_names:
name_to_username = _load_usernames_by_display_names(contact_db_path, missing_from_names)
if name_to_username:
for m in merged:
if str(m.get("fromUsername") or "").strip():
continue
if str(m.get("renderType") or "").strip() != "link":
continue
fn = str(m.get("from") or "").strip()
if fn and fn in name_to_username:
m["fromUsername"] = name_to_username[fn]
from_usernames = [str(m.get("fromUsername") or "").strip() for m in merged]
uniq_senders = list(
dict.fromkeys([u for u in (sender_usernames + list(pat_usernames) + quote_usernames) if u])
dict.fromkeys([u for u in (sender_usernames + list(pat_usernames) + quote_usernames + from_usernames) if u])
)
sender_contact_rows = _load_contact_rows(contact_db_path, uniq_senders)
local_sender_avatars = _query_head_image_usernames(head_image_db_path, uniq_senders)
for m in merged:
# If appmsg doesn't provide sourcedisplayname, try mapping sourceusername to display name.
if (not str(m.get("from") or "").strip()) and str(m.get("fromUsername") or "").strip():
fu = str(m.get("fromUsername") or "").strip()
frow = sender_contact_rows.get(fu)
if frow is not None:
m["from"] = _pick_display_name(frow, fu)
su = str(m.get("senderUsername") or "")
if not su:
continue
@@ -2479,6 +2516,8 @@ def _collect_chat_messages(
content_text = raw_text
title = ""
url = ""
from_name = ""
from_username = ""
record_item = ""
image_md5 = ""
emoji_md5 = ""
@@ -2523,6 +2562,8 @@ def _collect_chat_messages(
content_text = str(parsed.get("content") or "")
title = str(parsed.get("title") or "")
url = str(parsed.get("url") or "")
from_name = str(parsed.get("from") or "")
from_username = str(parsed.get("fromUsername") or "")
record_item = str(parsed.get("recordItem") or "")
quote_title = str(parsed.get("quoteTitle") or "")
quote_content = str(parsed.get("quoteContent") or "")
@@ -2725,6 +2766,7 @@ def _collect_chat_messages(
content_text = str(parsed.get("content") or content_text)
title = str(parsed.get("title") or title)
url = str(parsed.get("url") or url)
from_name = str(parsed.get("from") or from_name)
record_item = str(parsed.get("recordItem") or record_item)
quote_title = str(parsed.get("quoteTitle") or quote_title)
quote_content = str(parsed.get("quoteContent") or quote_content)
@@ -2785,6 +2827,8 @@ def _collect_chat_messages(
"content": content_text,
"title": title,
"url": url,
"from": from_name,
"fromUsername": from_username,
"recordItem": record_item,
"imageMd5": image_md5,
"imageFileId": image_file_id,
@@ -3124,6 +3168,8 @@ async def list_chat_messages(
content_text = raw_text
title = ""
url = ""
from_name = ""
from_username = ""
record_item = ""
image_md5 = ""
emoji_md5 = ""
@@ -3168,6 +3214,8 @@ async def list_chat_messages(
content_text = str(parsed.get("content") or "")
title = str(parsed.get("title") or "")
url = str(parsed.get("url") or "")
from_name = str(parsed.get("from") or "")
from_username = str(parsed.get("fromUsername") or "")
record_item = str(parsed.get("recordItem") or "")
quote_title = str(parsed.get("quoteTitle") or "")
quote_content = str(parsed.get("quoteContent") or "")
@@ -3366,6 +3414,7 @@ async def list_chat_messages(
content_text = str(parsed.get("content") or content_text)
title = str(parsed.get("title") or title)
url = str(parsed.get("url") or url)
from_name = str(parsed.get("from") or from_name)
record_item = str(parsed.get("recordItem") or record_item)
quote_title = str(parsed.get("quoteTitle") or quote_title)
quote_content = str(parsed.get("quoteContent") or quote_content)
@@ -3419,6 +3468,8 @@ async def list_chat_messages(
"content": content_text,
"title": title,
"url": url,
"from": from_name,
"fromUsername": from_username,
"recordItem": record_item,
"imageMd5": image_md5,
"imageFileId": image_file_id,
@@ -3546,15 +3597,44 @@ async def list_chat_messages(
is_sent = m.get("isSent", False)
m["transferStatus"] = "已收款" if is_sent else "已被接收"
# Some appmsg payloads provide only `from` (sourcedisplayname) but not `fromUsername` (sourceusername).
# Recover `fromUsername` via contact.db so the frontend can render the publisher avatar.
missing_from_names = [
str(m.get("from") or "").strip()
for m in merged
if str(m.get("renderType") or "").strip() == "link"
and str(m.get("from") or "").strip()
and not str(m.get("fromUsername") or "").strip()
]
if missing_from_names:
name_to_username = _load_usernames_by_display_names(contact_db_path, missing_from_names)
if name_to_username:
for m in merged:
if str(m.get("fromUsername") or "").strip():
continue
if str(m.get("renderType") or "").strip() != "link":
continue
fn = str(m.get("from") or "").strip()
if fn and fn in name_to_username:
m["fromUsername"] = name_to_username[fn]
from_usernames = [str(m.get("fromUsername") or "").strip() for m in merged]
uniq_senders = list(
dict.fromkeys(
[u for u in (sender_usernames + list(pat_usernames) + quote_usernames) if u]
[u for u in (sender_usernames + list(pat_usernames) + quote_usernames + from_usernames) if u]
)
)
sender_contact_rows = _load_contact_rows(contact_db_path, uniq_senders)
local_sender_avatars = _query_head_image_usernames(head_image_db_path, uniq_senders)
for m in merged:
# If appmsg doesn't provide sourcedisplayname, try mapping sourceusername to display name.
if (not str(m.get("from") or "").strip()) and str(m.get("fromUsername") or "").strip():
fu = str(m.get("fromUsername") or "").strip()
frow = sender_contact_rows.get(fu)
if frow is not None:
m["from"] = _pick_display_name(frow, fu)
su = str(m.get("senderUsername") or "")
if not su:
continue

View File

@@ -408,6 +408,91 @@ def _detect_media_type_and_ext(data: bytes) -> tuple[bytes, str, str]:
return payload, media_type, ext
def _is_allowed_proxy_image_host(host: str) -> bool:
"""Allowlist hosts for proxying images to avoid turning this into a general SSRF gadget."""
h = str(host or "").strip().lower()
if not h:
return False
# WeChat public account/article thumbnails and avatars commonly live on these CDNs.
return h.endswith(".qpic.cn") or h.endswith(".qlogo.cn")
@router.get("/api/chat/media/proxy_image", summary="代理获取远程图片(解决微信公众号图片防盗链)")
async def proxy_image(url: str):
u = html.unescape(str(url or "")).strip()
if not u:
raise HTTPException(status_code=400, detail="Missing url.")
if not _is_safe_http_url(u):
raise HTTPException(status_code=400, detail="Invalid url (only public http/https allowed).")
try:
p = urlparse(u)
except Exception:
raise HTTPException(status_code=400, detail="Invalid url.")
host = (p.hostname or "").strip().lower()
if not _is_allowed_proxy_image_host(host):
raise HTTPException(status_code=400, detail="Unsupported url host for proxy_image.")
def _download_bytes() -> tuple[bytes, str]:
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120 Safari/537.36",
"Accept": "image/avif,image/webp,image/apng,image/*,*/*;q=0.8",
# qpic/qlogo often require a mp.weixin.qq.com referer (anti-hotlink)
"Referer": "https://mp.weixin.qq.com/",
"Origin": "https://mp.weixin.qq.com",
}
r = requests.get(u, headers=headers, timeout=20, stream=True)
try:
r.raise_for_status()
content_type = str(r.headers.get("Content-Type") or "").strip()
max_bytes = 10 * 1024 * 1024
chunks: list[bytes] = []
total = 0
for ch in r.iter_content(chunk_size=64 * 1024):
if not ch:
continue
chunks.append(ch)
total += len(ch)
if total > max_bytes:
raise HTTPException(status_code=400, detail="Proxy image too large (>10MB).")
return b"".join(chunks), content_type
finally:
try:
r.close()
except Exception:
pass
try:
data, ct = await asyncio.to_thread(_download_bytes)
except HTTPException:
raise
except Exception as e:
logger.warning(f"proxy_image failed: url={u} err={e}")
raise HTTPException(status_code=502, detail=f"Proxy image failed: {e}")
if not data:
raise HTTPException(status_code=502, detail="Proxy returned empty body.")
payload, media_type, _ext = _detect_media_type_and_ext(data)
# Prefer upstream Content-Type when it looks like an image (sniffing may fail for some formats).
if media_type == "application/octet-stream" and ct:
try:
mt = ct.split(";")[0].strip()
if mt.startswith("image/"):
media_type = mt
except Exception:
pass
if not str(media_type or "").startswith("image/"):
raise HTTPException(status_code=502, detail="Proxy did not return an image.")
resp = Response(content=payload, media_type=media_type)
resp.headers["Cache-Control"] = "public, max-age=86400"
return resp
@router.post("/api/chat/media/emoji/download", summary="下载表情消息资源到本地 resource")
async def download_chat_emoji(req: EmojiDownloadRequest):
md5 = str(req.md5 or "").strip().lower()