Update app.py

app.py

@@ -1,4 +1,5 @@
-from asyncio import create_subprocess_shell, create_task, gather, sleep
+from asyncio import create_subprocess_shell, create_task, gather, run, sleep
+from json import dumps
 from logging import ERROR, INFO, basicConfig, getLogger
 from pathlib import Path
 from shutil import rmtree
@@ -7,22 +8,16 @@ from typing import Any, List
 from uuid import uuid4
 
 from PIL import Image
+from aiorentry.client import Client as RentryClient
 from fastapi import FastAPI, HTTPException
 from fastapi.responses import PlainTextResponse
 from httpx import AsyncClient, HTTPStatusError, RequestError
+from lxml import html
+from markdown import markdown
 from pydantic import BaseModel, HttpUrl
-from ytelegraph import TelegraphAPI, TelegraphAccount
-from aiorentry.client import Client as RentryClient
 from uvicorn import run as uvicorn_run
 
-from os import environ
-environ['PH_TOKEN_PATH'] = (Path(__file__).parent / 'ph.txt').as_posix()
-try:
-    ph = TelegraphAPI(access_token='886b4af0a626c532c0ea6e86cd3b362a32c09d6485cc8e2bccae956032cf')
-    ta = TelegraphAccount(access_token='886b4af0a626c532c0ea6e86cd3b362a32c09d6485cc8e2bccae956032cf')
-except:
-    ph = TelegraphAPI()
-    ta = TelegraphAccount()
+proxy_endpoint = 'https://telegraphprxver.vercel.app'
 
 need_logging = False
 
@@ -43,6 +38,112 @@ tokens = [
 ]
 
 
+class UserInfo:
+    def __init__(self, token: str, short_name: str, author_name: str, author_url: str, auth_link: str):
+        self.token = token
+        self.short_name = short_name
+        self.author_name = author_name
+        self.author_url = author_url
+        self.auth_link = auth_link
+
+    def get_token(self) -> str:
+        return self.token
+
+    def get_short_name(self) -> str:
+        return self.short_name
+
+    def get_author_name(self) -> str:
+        return self.author_name
+
+    def get_author_url(self) -> str:
+        return self.author_url
+
+    def get_auth_link(self) -> str:
+        return self.auth_link
+
+    def __repr__(self):
+        return f'UserAuthInfo(token={self.token}, short_name={self.short_name}, author_name={self.author_name}, author_url={self.author_url}, auth_link={self.auth_link})'
+
+
+async def create_account(short_name: str, author_name: str, author_url: str) -> UserInfo:
+    params = {
+        'short_name': short_name,
+        'author_name': author_name,
+        'author_url': author_url
+    }
+    async with AsyncClient(verify=False, follow_redirects=True, timeout=30.0) as client:
+        response = await client.get(f'{proxy_endpoint}/createAccount', params=params)
+        json_response: dict = response.json()
+        if json_response.get('ok'):
+            result = json_response.get('result', {})
+            return UserInfo(result['access_token'], result['short_name'], result['author_name'], result['author_url'], result['auth_url'])
+
+
+tgph_acc = run(create_account('prolapse', 'prolapse', ''))
+
+
+def md_to_dom(markdown_text: str) -> list[dict[str, str | list | dict | None]]:
+    def handle_heading(element) -> dict[str, str | list | dict | None]:
+        if element.tag == 'h1':
+            return {'tag': 'h3', 'children': parse_children(element)}
+        elif element.tag == 'h2':
+            return {'tag': 'h4', 'children': parse_children(element)}
+        else:
+            return {'tag': 'p', 'children': [{'tag': 'strong', 'children': parse_children(element)}]}
+
+    def handle_list(element) -> dict[str, str | list | dict | None]:
+        return {'tag': element.tag, 'children': [{'tag': 'li', 'children': parse_children(li)} for li in element.xpath('./li')]}
+
+    def handle_link(element) -> dict[str, str | list | dict | None]:
+        return {'tag': 'a', 'attrs': {'href': element.get('href')}, 'children': parse_children(element)}
+
+    def handle_media(element) -> dict[str, str | list | dict | None]:
+        return {'tag': element.tag, 'attrs': {'src': element.get('src')}}
+
+    def parse_children(element) -> list[str | dict[str, str | list | dict | None]]:
+        return [parse_element(child) if child.tag else child.strip() for child in element.iterchildren() if child.tag or (isinstance(child, str) and child.strip())]
+
+    def parse_element(element) -> dict[str, str | list | dict | None]:
+        handlers = {'h1': handle_heading, 'h2': handle_heading, 'h3': handle_heading, 'h4': handle_heading, 'h5': handle_heading, 'h6': handle_heading, 'ul': handle_list, 'ol': handle_list, 'a': handle_link, 'img': handle_media, 'iframe': handle_media}
+        handler = handlers.get(element.tag, lambda e: {'tag': e.tag, 'children': parse_children(e)})
+        return handler(element)
+
+    html_content = markdown(markdown_text, extensions=['extra', 'sane_lists'])
+    tree = html.fromstring(html_content)
+    return [parse_element(element) for element in tree.body]
+
+
+async def tgph_create_page(token: str, title: str, markdown_text: str) -> str:
+    params = {
+        'access_token': token,
+        'title': title,
+        'content': dumps(md_to_dom(markdown_text)),
+        'return_content': False
+    }
+    async with AsyncClient(verify=False, follow_redirects=True, timeout=30.0) as client:
+        response = await client.get(f'{proxy_endpoint}/createPage', params=params)
+        json_response: dict = response.json()
+        if json_response.get('ok'):
+            result = json_response.get('result', {})
+        return result.get('path')
+
+
+async def tgph_edit_page(token: str, page: str, title: str, markdown_text: str) -> str:
+    params = {
+        'access_token': token,
+        'path': page,
+        'title': title,
+        'content': dumps(md_to_dom(markdown_text)),
+        'return_content': False
+    }
+    async with AsyncClient(verify=False, follow_redirects=True, timeout=30.0) as client:
+        response = await client.get(f'{proxy_endpoint}/editPage', params=params)
+        json_response = response.json()
+        if json_response.get('ok'):
+            result = json_response.get('result', {})
+        return result.get('path')
+
+
 async def download_png(url: str, folder: str, client: AsyncClient, retries: int = 5) -> Path:
     # print(f'загрузка изображения: {url}')
     for attempt in range(retries):
@@ -178,7 +279,7 @@ async def optimize_and_upload(images_urls: List[str] | str, convert: bool = Fals
     images_urls = [images_urls] if isinstance(images_urls, str) else images_urls
     if convert:
         try:
-            ph_link = ph.create_page_md('DAll-E v3', 'изображения скоро появятся, обнови страницу...')
+            ph_link = tgph_create_page(tgph_acc.get_token(), 'DAll-E v3', '*изображения скоро появятся, обнови страницу...*')
         except Exception as e:
             print(f'не получилось создать страницу на телеграфе: {e}')
             ph_link = None
@@ -186,8 +287,8 @@ async def optimize_and_upload(images_urls: List[str] | str, convert: bool = Fals
             page = await client.new_page('изображения скоро появятся, обнови страницу...')
             page_id, code = page.url, page.edit_code
             continue_task = create_task(continue_optimizing_and_uploading(images_urls, page_id, code, ph_link))
-            print(f'https://rentry.co/{page_id}', f'https://rentry.org/{page_id}')
-            return [ph_link, f'https://rentry.co/{page_id}', f'https://rentry.org/{page_id}'] if ph_link else [f'https://rentry.co/{page_id}', f'https://rentry.org/{page_id}']
+            print(f'https://telegra.ph/{ph_link}', f'https://rentry.co/{page_id}', f'https://rentry.org/{page_id}')
+            return [f'https://telegra.ph/{ph_link}', f'https://rentry.co/{page_id}', f'https://rentry.org/{page_id}'] if ph_link else [f'https://rentry.co/{page_id}', f'https://rentry.org/{page_id}']
     else:
         return await continue_optimizing_and_uploading(images_urls)
 
@@ -217,17 +318,13 @@ async def continue_optimizing_and_uploading(images_urls: list[str], page_id: str
     content = '\n\n'.join([f'![]({url})' for url in new_images_urls])
     if ph_link:
         try:
-            ph.edit_page_md(ph_link, content)
+            await tgph_edit_page(tgph_acc.get_token(), ph_link, 'DAll-E v3', content)
         except Exception as e:
-            print(f'не удалось создать страницу на телеграфе: {e}')
+            print(f'не удалось создать отредактировать на телеграфе: {e}')
     if page_id and code:
         try:
             async with RentryClient('https://rentry.org') as client:
-                await client.edit_page(
-                    text=content,
-                    url=page_id,
-                    edit_code=code,
-                )
+                await client.edit_page(text=content, url=page_id, edit_code=code)
         except Exception as e:
             print(f'не удалось создать страницу в rentry: {e}')
 
@@ -266,4 +363,3 @@ async def optimize_images_endpoint(image_urls: ImageURLs):
 
 if __name__ == "__main__":
     uvicorn_run(app, host='0.0.0.0', port=7860, timeout_keep_alive=90, log_level='info', use_colors=False)
-