Telegram-Bot-API: Senden Sie Videos ohne Komprimierung und vermeiden Sie die Vorschau schwarzer Quadrate (keine Miniatur ⇐ Python

[Anonymous]

Ich verwende Python mit aiohttp, um Videos aus der Facebook-Werbebibliothek über die Bot-API an Telegram zu senden. Ich stehe vor zwei Hauptproblemen:

• Videokomprimierung: Telegram komprimiert die Videos automatisch, was die Qualität verringert. Ich möchte sie als „Streamable Videos“ senden, aber eine hohe Qualität beibehalten oder verstehen, wie man sendDocument für diesen Zweck richtig verwendet.
• Vorschau des schwarzen Quadrats: Wenn das Video gesendet wird, erscheint es im Chat als schwarzes Quadrat, bis es abgespielt wird. Ich rufe Video-URLs direkt vom CDN von Facebook ab.

Aktuelle Logik (vereinfacht): Ich extrahiere gerade video_hd_url und sende sie. Allerdings weiß ich nicht, wie ich programmgesteuert ein Miniaturbild generieren oder anhängen kann, um den schwarzen Platzhalter zu vermeiden.

Code: Select all

async def send_ad_to_telegram(
bot_token: str,
session: aiohttp.ClientSession,
ad: dict,
duplicate_count: int = 0,
message_thread_id: str = None,
custom_geo_string: str = None
):
snapshot = ad.get("snapshot", {})
ad_id = ad.get("ad_archive_id")

# ==========================================
# 1. GEO AND TARGETING LOGIC
# ==========================================

# 1.1 Get list of countries (Target Geo)
geos_set = set()
raw_targets = ad.get("target_locations")
if raw_targets:
parts = [x.strip() for x in raw_targets.split(",") if x.strip()]
geos_set.update(parts)

# If empty, try to extract from demographic data
if not geos_set:
try:
demo = ad.get("demographic_data", {})
breakdown = demo.get("age_gender_breakdown", [])
for item in breakdown:
c = item.get("country")
if c: geos_set.add(c)
if not geos_set:
locs = demo.get("location_audience", [])
for item in locs:
name = item.get("name")
if name: geos_set.add(name)
except: pass

# 1.2 Target string for body: "#AT, #DE"
if geos_set:
target_geo_list = " ".join([f"#{g}" for g in sorted(list(geos_set))])
else:
target_geo_list = None

# 1.3 🔥 MAIN HASHTAG (STRICTLY: COUNTRY OR EU) 🔥
header_geo = REVERSE_TOPIC_MAP.get(message_thread_id)

if not header_geo:
# If single country -> use it
if len(geos_set) == 1:
header_geo = list(geos_set)[0]
# If multiple countries -> EU
else:
# Try to extract from ad text
extracted = extract_geo_from_ad(ad)

if extracted and str(extracted).upper() not in ["N/A", "UNKNOWN", "NONE", ""]:
header_geo = extracted
else:
# 🔥 FALLBACK: If nothing found - set to EU
header_geo = "EU"

# ==========================================
# 2.  AD DATA
# ==========================================
first_card = (snapshot.get("cards") or [{}])[0]
page_name = snapshot.get("page_name", "No Name")
is_active = ad.get("is_active", None)
page_url = snapshot.get("page_profile_uri", "")

body_text = first_card.get("body") or (snapshot.get("body") or {}).get("text", "")
body = clean_ad_text(body_text, max_lines=10)

link_desc_text = first_card.get("link_description") or snapshot.get("link_description", "")
link_description = clean_ad_text(link_desc_text)

link = first_card.get("link_url") or snapshot.get("link_url", "")
cta = first_card.get("cta_text") or snapshot.get("cta_text", "Learn More")
domain = get_domain(link)

platforms_list = ad.get("publisher_platform", [])
platforms = abbreviate_platforms(platforms_list)

page_like = (ad.get("snapshot") or {}).get("page_like_count") or \
((ad.get("snapshot") or {}).get("body") or {}).get("page_like_count") or \
(((ad.get("advertiser") or {}).get("ad_library_page_info") or {}).get("page_info") or {}).get("likes") or \
(ad.get("body") or {}).get("page_like_count") or 0

start_date = format_start_date_new(ad.get("start_date_formatted"))
active_time = format_active_time(ad.get("total_active_time"))

reach = (ad.get("demographic_data") or {}).get("eu_total_reach")
if not reach:
reach = (ad.get("eu_transparency") or {}).get("eu_total_reach")

platform_tags = get_platform_tags(link)
if not platform_tags:
return False

ages = ad.get("age_range_overall")

ad_library_url = ad.get("ad_library_url") or (f"https://www.facebook.com/ads/library/?id={ad_id}" if ad_id else "")
source_search_url = ad.get("url")

# --- MEDIA ---
video_url, image_url = None, None
cards = snapshot.get("cards", [])
videos = snapshot.get("videos", [])
images = snapshot.get("images", [])

if isinstance(cards, list) and cards:
for card in cards:
if card.get("video_hd_url") or card.get("video_sd_url"):
video_url = card.get("video_hd_url") or card.get("video_sd_url")
break
if not video_url and isinstance(videos, list) and videos:
video_url = videos[0].get("video_hd_url") or videos[0].get("video_sd_url")
if not video_url:
if isinstance(cards, list) and cards:
for card in cards:
if card.get("original_image_url") or card.get("resized_image_url"):
image_url = card.get("original_image_url") or card.get("resized_image_url")
break
if not image_url and isinstance(images, list) and images:
image_url = images[0].get("original_image_url") or images[0].get("resized_image_url")

media_tag = get_media_type_tag(video_url, image_url)

# --- MESSAGE ASSEMBLY ---
if custom_geo_string:
header_geo = False
geo_line_display = f"Geo (tag):[/b] {custom_geo_string}"[b]        if custom_geo_string not in ["#WW", "#EU"] and len(custom_geo_string.split()) == 1:
return False
main_tag = ""
else:
geo_line_display = None
main_tag = f"#{header_geo}"

tags_line = " ".join(platform_tags)

def build_message(include_body=True, include_desc=True):
carousel ="#CAROUSEL"

if custom_geo_string:
full_tags_line = f"{tags_line} {carousel if len(cards)>=2 else ''} {media_tag}"
else:
full_tags_line = f"{main_tag} {tags_line} {carousel if len(cards)>=2 else ''} {media_tag}"

full_tags_line = " ".join(full_tags_line.split())

activity_line = f"Active:[/b] {'Yes' if is_active else 'No'}"  if is_active is not None else None[b]
parts = []
parts.append(f"{full_tags_line} {start_date}[/b]")[b]
if target_geo_list:
parts.append(f"Geo:[/b] {target_geo_list}")[b]        elif geo_line_display:
parts.append(geo_line_display)
if ages: parts.append(f"Age:[/b] {ages}")[b]        if active_time: parts.append(f"Total active time:[/b] {active_time}")[b]        if activity_line: parts.append(activity_line)

links_block = []
if ad_library_url: links_block.append(f"Library:[/b] [url={ad_library_url}]{ad_id}[/url]")[b]        if links_block: parts.append(" | ".join(links_block))

if page_url: parts.append(f"Page:[/b] [url={page_url}]{page_name}[/url] - {page_like}👥")[b]        parts.append(f"Platforms:[/b] {platforms}")[b]
current_body = clean_ad_text(body) if include_body else ""
current_desc = clean_ad_text(link_description) if include_desc else ""

if current_body: parts.append(f"Text:[/b]\n[b]{truncate_text(current_body, BODY_TEXT_LIMIT)}")
if current_desc: parts.append(f"Description:[/b]\n[b]{truncate_text(current_desc, BODY_TEXT_LIMIT)}")

app_id = get_app_id_from_url(link)
link_display = app_id if ("#WebView" in platform_tags or "#iOS" in platform_tags) and app_id else domain

if link: parts.append(f"Link:[/b] [url={link}]{link_display or 'Link'}[/url]")[b]        if cta: parts.append(f"CTA:[/b] {cta}")[b]
if duplicate_count > 0:
suffix = "🔹" if duplicate_count  TELEGRAM_CAPTION_LIMIT: message = build_message(True, False)
if len(message) > TELEGRAM_CAPTION_LIMIT: message = build_message(False, False)
if len(message) > TELEGRAM_CAPTION_LIMIT: message = truncate_text(message, TELEGRAM_CAPTION_LIMIT)

temp_files = []