feat: init

.gitignore

@@ -0,0 +1,5 @@
+__pycache__
+.venv
+.vscode
+*.session
+*.toml

TgFileSystemClient.py

@@ -0,0 +1,51 @@
+import asyncio
+from typing import Union
+
+from telethon import TelegramClient, types
+
+import configParse
+
+
+class TgFileSystemClient(object):
+    api_id: int
+    api_hash: str
+    session_name: str
+    proxy_param: dict[str, any]
+    client: TelegramClient
+    me: Union[types.User, types.InputPeerUser]
+
+    def __init__(self, param: configParse.TgToFileSystemParameter) -> None:
+        self.api_id = param.tgApi.api_id
+        self.api_hash = param.tgApi.api_hash
+        self.session_name = param.base.name
+        self.proxy_param = {
+            'proxy_type': param.proxy.proxy_type,
+            'addr': param.proxy.addr,
+            'port': param.proxy.port,
+        } if param.proxy.enable else {}
+        self.client = TelegramClient(
+            self.session_name, self.api_id, self.api_hash, proxy=self.proxy_param)
+
+
+    def __repr__(self) -> str:
+        if not self.client.is_connected:
+            return f"client disconnected, session_name:{self.session_name}"
+        return f"client connected, session_name:{self.session_name}, username:{self.me.username}, phone:{self.me.phone}, detail:{self.me.stringify()}"
+
+    async def init_client(self):
+        self.me = await self.client.get_me()
+
+    def __enter__(self):
+        self.client.__enter__()
+        self.client.loop.run_until_complete(self.init_client())
+
+    def __exit__(self):
+        self.client.__exit__()
+        self.me = None
+
+    async def __aenter__(self):
+        await self.client.__enter__()
+        await self.init_client()
+
+    async def __aexit__(self):
+        await self.client.__aexit__()

TgFileSystemClientManager.py

@@ -0,0 +1,35 @@
+from typing import Any
+from TgFileSystemClient import TgFileSystemClient
+
+
+class TgFileSystemClientManager(object):
+    MAX_MANAGE_CLIENTS: int = 10
+    clients: dict[int, TgFileSystemClient]
+    
+    def __init__(self) -> None:
+        pass
+    
+    def push_client(self, client: TgFileSystemClient) -> int:
+        """
+        push client to manager.
+
+        Arguments
+            client
+
+        Returns
+            client id
+
+        """
+        self.clients[id(client)] = client
+        return id(client)
+        
+    def get_client(self, client_id: int) -> TgFileSystemClient:
+        client = self.clients.get(client_id)
+        return client
+    
+    
+
+if __name__ == "__main__":
+    import configParse
+    t: TgFileSystemClient = TgFileSystemClient(configParse.get_TgToFileSystemParameter())
+    print(f"{t.session_name=}")

configParse.py

@@ -0,0 +1,29 @@
+import toml
+from pydantic import BaseModel
+
+
+class TgToFileSystemParameter(BaseModel):
+    class BaseParameter(BaseModel):
+        name: str
+        port: int
+    base: BaseParameter
+
+    class ApiParameter(BaseModel):
+        api_id: int
+        api_hash: str
+    tgApi: ApiParameter
+
+    class TgProxyParameter(BaseModel):
+        enable: bool
+        proxy_type: str
+        addr: str
+        port: int
+    proxy: TgProxyParameter
+
+__cache_res = None
+def get_TgToFileSystemParameter(path: str = "./config.toml", force_reload: bool = False) -> TgToFileSystemParameter:
+    global __cache_res
+    if __cache_res is not None and not force_reload:
+        return __cache_res
+    __cache_res = TgToFileSystemParameter.model_validate(toml.load(path))
+    return __cache_res

requirements.txt

@@ -0,0 +1,5 @@
+toml
+telethon
+# python-socks[asyncio]
+fastapi
+uvicorn[standard]

start.py

@@ -0,0 +1,76 @@
+import asyncio
+
+import uvicorn
+from fastapi import FastAPI
+from fastapi import status
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import Response
+from contextlib import asynccontextmanager
+from telethon import TelegramClient
+
+import configParse
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    param = configParse.get_TgToFileSystemParameter()
+    loop = asyncio.get_event_loop()
+    tg_client_task = loop.create_task(start_tg_client(param))
+    yield
+    asyncio.gather(*[tg_client_task])
+
+app = FastAPI(lifespan=lifespan)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.post("/tg/{chat_id}/{message_id}")
+async def get_test(chat_id: str, message_id: str):
+    print(f"test: {chat_id=}, {message_id=}")
+    return Response(status_code=status.HTTP_200_OK)
+    
+
+async def start_tg_client(param: configParse.TgToFileSystemParameter):
+    api_id = param.tgApi.api_id
+    api_hash = param.tgApi.api_hash
+    session_name = param.base.name
+    proxy_param = {
+        'proxy_type': param.proxy.proxy_type,
+        'addr': param.proxy.addr,
+        'port': param.proxy.port,
+    } if param.proxy.enable else {}
+    client = TelegramClient(session_name, api_id, api_hash, proxy=proxy_param)
+
+    async def tg_client_main():
+        # Getting information about yourself
+        me = await client.get_me()
+
+        # "me" is a user object. You can pretty-print
+        # any Telegram object with the "stringify" method:
+        print(me.stringify())
+
+        # When you print something, you see a representation of it.
+        # You can access all attributes of Telegram objects with
+        # the dot operator. For example, to get the username:
+        username = me.username
+        print(username)
+        print(me.phone)
+        # You can print all the dialogs/conversations that you are part of:
+        dialogs = await client.get_dialogs()
+        for dialog in dialogs:
+            print(f"{dialog.name} has ID {dialog.id}")
+        # async for dialog in client.iter_dialogs():
+        #     print(dialog.name, 'has ID', dialog.id)
+
+    async with client:
+        await tg_client_main()
+
+
+
+if __name__ == "__main__":
+    param = configParse.get_TgToFileSystemParameter()
+    uvicorn.run(app, host="0.0.0.0", port=param.base.port)

test.py

@@ -0,0 +1,74 @@
+from telethon import TelegramClient
+
+import configParse
+
+param = configParse.get_TgToFileSystemParameter()
+# Remember to use your own values from my.telegram.org!
+api_id = param.ApiParameter.api_id
+api_hash = param.ApiParameter.api_hash
+client = TelegramClient('anon', api_id, api_hash, proxy={
+    'proxy_type': 'socks5',
+    'addr': '172.25.32.1',
+    'port': 7890,
+})
+# client = TelegramClient('anon', api_id, api_hash, proxy=("socks5", '127.0.0.1', 7890))
+# proxy=("socks5", '127.0.0.1', 4444)
+
+async def main():
+    # Getting information about yourself
+    me = await client.get_me()
+
+    # "me" is a user object. You can pretty-print
+    # any Telegram object with the "stringify" method:
+    print(me.stringify())
+
+    # When you print something, you see a representation of it.
+    # You can access all attributes of Telegram objects with
+    # the dot operator. For example, to get the username:
+    username = me.username
+    print(username)
+    print(me.phone)
+
+    # You can print all the dialogs/conversations that you are part of:
+    async for dialog in client.iter_dialogs():
+        print(dialog.name, 'has ID', dialog.id)
+
+    # You can send messages to yourself...
+    # await client.send_message('me', 'Hello, myself!')
+    # ...to some chat ID
+    # await client.send_message(-100123456, 'Hello, group!')
+    # ...to your contacts
+    # await client.send_message('+34600123123', 'Hello, friend!')
+    # ...or even to any username
+    # await client.send_message('username', 'Testing Telethon!')
+
+    # You can, of course, use markdown in your messages:
+    # message = await client.send_message(
+    #     'me',
+    #     'This message has **bold**, `code`, __italics__ and '
+    #     'a [nice website](https://example.com)!',
+    #     link_preview=False
+    # )
+
+    # Sending a message returns the sent message object, which you can use
+    # print(message.raw_text)
+
+    # You can reply to messages directly if you have a message object
+    # await message.reply('Cool!')
+
+    # Or send files, songs, documents, albums...
+    # await client.send_file('me', './test.py')
+
+    # You can print the message history of any chat:
+    message = await client.get_messages('me', ids=206963)
+    async for message in client.iter_messages('me'):
+        print(message.id, message.text)
+
+        # You can download media from messages, too!
+        # The method will return the path where the file was saved.
+        # if message.photo:
+        #     path = await message.download_media()
+        #     print('File saved to', path)  # printed after download is done
+
+with client:
+    client.loop.run_until_complete(main())