Telegram Bot API: отправляйте видео без сжатия и избегайте предварительного просмотра в виде черного квадрата (без миниа ⇐ Python

[Anonymous]

Я использую python с aiohttp для отправки видео из библиотеки рекламы Facebook в Telegram через Bot API. Я столкнулся с двумя основными проблемами:

• Сжатие видео: Telegram автоматически сжимает видео, что снижает качество. Я хочу отправлять их как «Потоковое видео», но с сохранением высокого качества, или понимать, как правильно использовать sendDocument для этой цели.
• Предварительный просмотр черного квадрата: Когда видео отправляется, оно отображается в чате в виде черного квадрата, пока не будет воспроизведено. Я получаю URL-адреса видео непосредственно из CDN Facebook.

Текущая логика (упрощенная): В настоящее время я извлекаю video_hd_url и отправляю его. Однако я не знаю, как программно создать или прикрепить миниатюру, чтобы избежать черного заполнителя.
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): {custom_geo_string}" 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: {'Yes' if is_active else 'No'}" if is_active is not None else None
parts = []
parts.append(f"{full_tags_line} {start_date}")
if target_geo_list:
parts.append(f"Geo: {target_geo_list}") elif geo_line_display:
parts.append(geo_line_display)
if ages: parts.append(f"Age: {ages}") if active_time: parts.append(f"Total active time: {active_time}") if activity_line: parts.append(activity_line)

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

if page_url: parts.append(f"Page: {page_name} - {page_like}") parts.append(f"Platforms: {platforms}")
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:\n{truncate_text(current_body, BODY_TEXT_LIMIT)}")
if current_desc: parts.append(f"Description:\n{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: {link_display or 'Link'}") if cta: parts.append(f"CTA: {cta}")
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 = []


Подробнее здесь: https://stackoverflow.com/questions/798 ... re-preview