Commit e52f565
Changed files (4)
src
src/danmu/entrypoint.py
@@ -127,7 +127,7 @@ async def query_danmu(client: Client, message: Message, **kwargs):
# less than 200K, add instant view
if file_bytes(merged_paths) < 204800:
texts = "\n".join([Path(path).read_text() for path in sorted(merged_paths, reverse=True)]).strip()
- if telegraph_url := await publish_telegraph(title=f"【{qtype}】{user}{match_time} {keyword}", html=convert_html(texts), author=user):
+ if telegraph_url := await publish_telegraph(title=f"【{qtype}】{user}{match_time} {keyword}", html=convert_html(texts), author=user, ttl="1d"):
caption += f"\n⚡️[即时预览]({telegraph_url})"
await send2tg(client, message, texts=caption, media=media, **kwargs)
[Path(path).unlink(missing_ok=True) for path in paths]
src/history/query.py
@@ -118,7 +118,14 @@ async def query_chat_history(client: Client, message: Message, **kwargs):
texts = results.get("full_texts", "") # use full texts
caption += f"\n#️⃣消息数: {count}"
# less than 100,000, add instant view
- if len(texts) < 1000000 and (telegraph_url := await publish_telegraph(title=f"【{chat_title}】{user}{match_time} {keyword}", html=convert_html(texts), author=user or chat_title)):
+ if len(texts) < 1000000 and (
+ telegraph_url := await publish_telegraph(
+ title=f"【{chat_title}】{user}{match_time} {keyword}",
+ html=convert_html(texts),
+ author=user or chat_title,
+ ttl="1d",
+ )
+ ):
caption += f"\n⚡️[即时预览]({telegraph_url})"
# send as txt
with BytesIO(texts.encode("utf-8")) as f:
src/others/tmdb.py
@@ -305,7 +305,7 @@ async def get_people_details(people_id: int, tmdb_lang: Literal["en-US", "zh-CN"
images = sorted(images, key=lambda x: x.get("width", 0), reverse=True)[:10] # kepp 10 images
media = [{"photo": download_file(f"https://image.tmdb.org/t/p/original{img.get('file_path')}", proxy=PROXY.TMDB)} for img in images]
media = await download_media(media)
- telegraph_url = await publish_telegraph(title=name, html=productions_for_html.strip("<br>"), author=name, url=f"https://www.themoviedb.org/person/{people_id}")
+ telegraph_url = await publish_telegraph(title=name, html=productions_for_html.strip("<br>"), author=name, url=f"https://www.themoviedb.org/person/{people_id}", ttl="1d")
description = f"简介: {zhcn(resp['biography'])}" if resp.get("biography") else ""
description = description.replace("\\n", "\n")
src/publish.py
@@ -4,6 +4,7 @@ import contextlib
import io
import tempfile
from pathlib import Path
+from typing import Literal
from urllib.parse import quote_plus
import anyio
@@ -18,7 +19,14 @@ from database.r2 import set_cf_r2
from utils import nowdt, rand_string
-async def publish_telegraph(title: str, texts: str | None = None, html: str = "", author: str | None = None, url: str | None = None) -> str:
+async def publish_telegraph(
+ title: str,
+ texts: str | None = None,
+ html: str = "",
+ author: str | None = None,
+ url: str | None = None,
+ ttl: Literal["forever", "1d", "7d", "30d"] = "forever",
+) -> str:
"""Publish to Telegraph."""
def clean_html(s: str | None) -> str:
@@ -53,10 +61,17 @@ async def publish_telegraph(title: str, texts: str | None = None, html: str = ""
return page["url"]
except Exception as e:
logger.error(f"Telegraph publish error: {e}")
- return await publish_cf_r2(title, texts=texts, html=html, author=author, url=url)
+ return await publish_cf_r2(title, texts=texts, html=html, author=author, url=url, ttl=ttl)
-async def publish_cf_r2(title: str, texts: str | None = None, html: str = "", author: str | None = None, url: str | None = None) -> str:
+async def publish_cf_r2(
+ title: str,
+ texts: str | None = None,
+ html: str = "",
+ author: str | None = None,
+ url: str | None = None,
+ ttl: Literal["forever", "1d", "7d", "30d"] = "forever",
+) -> str:
"""Publish to CF R2."""
if not (texts or html):
return ""
@@ -64,7 +79,7 @@ async def publish_cf_r2(title: str, texts: str | None = None, html: str = "", au
html = markdown.markdown(texts)
now = nowdt(TZ)
today = f"{now:%Y-%m-%d}"
- key = f"InstantView/{today}-{rand_string(8)}.html"
+ key = f"InstantView/{today}-{rand_string(8)}.html" if ttl == "forever" else f"TTL/{ttl}/{today}-{rand_string(8)}.html"
if not url:
url = "https://instantview.telegram.org"
if not author: