# `vk.py` generate from `vk.ipynb` by `build.ipynb` # VK #ignore import os def load_env(file_path): """Загружает переменные окружения из файла env с поддержкой булевых значений.""" with open(file_path) as f: for line in f: line = line.strip() if not line or line.startswith('#'): continue try: key, value = line.split('=', 1) key = key.strip() value = value.strip('"\' ') # Преобразование строковых значений в булевые if value.lower() in ['true', '1', 'yes']: value = True elif value.lower() in ['false', '0', 'no']: value = False # Проверка, что ключ не пустой if key: os.environ[key] = str(value) # Храним как строку except ValueError: print(f"Ошибка: неверный формат в строке '{line}'") try: load_env('env') except Exception: # env_example тоже может отсутствовать (в образ эти файлы не копируются) — # тогда переменные берутся из окружения контейнера/k8s-секрета try: load_env('env_example') except Exception: pass ## Class import httpx import asyncio from db import * from config import * from auth import * from token_manager import TokenManager import logging from PIL import Image from io import BytesIO from celery import Celery from bson import ObjectId from fastapi import status from fastapi import APIRouter, Depends, Request, Response from fastapi.responses import RedirectResponse from images import insert_logo ### Конфигурация celery_app_vk = Celery('vk', broker=f'{redis_uri}1', backend=f'{redis_uri}1', task_track_started=True ) token_manager = TokenManager() router = APIRouter(prefix="/vk", tags=["VK OAuth"]) @router.get("/auth") async def auth_redirect( user=Depends(get_current_user) ): url = await token_manager.get_auth_url(user_id=user['id']) return RedirectResponse(url=url, status_code=status.HTTP_303_SEE_OTHER) @router.get("/register") async def vk_register_start(next: str = '/profile'): """Старт регистрации/входа через VK (без предварительной авторизации).""" url = await token_manager.get_auth_url(purpose='register', next=next) return RedirectResponse(url=url, status_code=status.HTTP_303_SEE_OTHER) @router.get("/token") async def oauth_callback(request: Request, response: Response, code: str, state: str, device_id: str = None): result = await token_manager.get_token(code=code, state=state, device_id=device_id) if not result or not result.get('user_id'): return RedirectResponse(url='/login', status_code=status.HTTP_303_SEE_OTHER) # Регистрация/вход через VK — выдаём нашу сессию и редиректим на next if result.get('purpose') == 'register': await issue_session(result['user_id'], request, response, client_id='vk') next_url = result.get('next') or '/profile' if not is_safe_redirect(next_url): next_url = '/profile' return RedirectResponse(url=next_url, status_code=status.HTTP_303_SEE_OTHER, headers=response.headers) # Привязка соцсети к уже залогиненному пользователю return RedirectResponse(url='/profile/social', status_code=status.HTTP_303_SEE_OTHER) async def get_images(db_name: str, file_ids: list, sign_avatar: bool = True): images = {} photo_bytes_list= [] if sign_avatar: user = await users_collection.find_one({'_id': ObjectId(file_ids[0])},{'img':1}) stream = await gridfs_map['users'].open_download_stream(ObjectId(user['img'])) avatar_bytes = await stream.read() logo = Image.open(BytesIO(avatar_bytes)).convert("RGBA") file_ids = file_ids[1:] for _id in file_ids: stream = await gridfs_map[db_name].open_download_stream(ObjectId(_id)) photo_bytes_list.append((stream.name, BytesIO(await stream.read()), _id)) for name, background, _id in photo_bytes_list: if sign_avatar: background = insert_logo(Image.open(background).convert("RGBA"), logo) images[_id] = ("photo", (name, background, "image/jpeg")) return images @router.post("/send") async def send_vk(data:dict, request: Request): auth = request.headers.get("Authorization") if auth and auth.lower().startswith("bearer "): parts = auth.split(" ", 1) if len(parts) == 2: if parts[1].strip() != ATOKEN: return None item = data['item'] text = data['text'] file_ids = data['file_ids'] doc = data['doc'] sign_avatar = data['sign_avatar'] db_name = item.get('db') # if db_name == 'competitions': # social = await social_collection.find_one({'service': True, 'type':'vk'}) # else: social = doc logging.info(doc) logging.info(social) images = await get_images(db_name, file_ids, sign_avatar) async with httpx.AsyncClient(http2=True, verify=ctx) as client: attachments = [] for _id, img in images.items(): # doc_photo = await mongo_client['vk']['photos'].find_one({'id':f"{_id}:{social.get('_id')}"}) # if doc_photo: # attachments.append(doc_photo['attachment']) # continue payload = {"v": "5.199"} if 'group' in social: payload['group_id'] = social['id'] resp = await client.post(f"{VK_API_URI}photos.getWallUploadServer", headers={"Authorization": f"Bearer {await token_manager.get_access_token(social['user_id'])}"}, data=payload, timeout=30.0) resp.raise_for_status() upload_url = resp.json()['response']['upload_url'] resp = await client.post(upload_url, files=[img], timeout=60.0) resp.raise_for_status() payload.update(resp.json()) payload.update({'caption':text}) resp = await client.post(f"{VK_API_URI}photos.saveWallPhoto", headers={"Authorization": f"Bearer {await token_manager.get_access_token(social['user_id'])}"}, data=payload, timeout=30.0) resp.raise_for_status() attachment = ','.join([f'photo{i['owner_id']}_{i['id']}' for i in resp.json()['response']]) if not attachment: return {'ok': False} # await mongo_client['vk']['photos'].insert_one({'attachment':attachment,'id':f"{_id}:{social.get('_id')}"}) attachments.append(attachment) logging.info(attachments) resp = await client.post(f"{VK_API_URI}wall.post", data={ "access_token": await token_manager.get_access_token(doc['user_id']), "owner_id": f'-{doc['id']}' if 'group' in doc else doc['id'], 'message': text, 'attachments': ','.join(attachments), 'primary_attachments_mode': 'grid', "v": "5.199", }) res = resp.json() res.update({'doc': doc}) if 'error' in res: return {'ok': False} update_spec = {"$push": {"vk": res}} await mongo_client[db_name][item.get('col')].update_one({'_id': ObjectId(item.get('_id'))}, update_spec) return {"ok": True} @router.post("/edit") async def edit_vk(data:dict, request: Request): auth = request.headers.get("Authorization") if auth and auth.lower().startswith("bearer "): parts = auth.split(" ", 1) if len(parts) == 2: if parts[1].strip() != ATOKEN: return None item = data['item'] text = data['text'] file_ids = data['file_ids'] doc = data['doc'] post_id = data['post_id'] sign_avatar = data['sign_avatar'] # async def edit_vk(item: dict, text: str, file_ids: list, doc: dict, sign_avatar: bool = True): db_name = item.get('db') if db_name == 'competitions': social = await social_collection.find_one({'service': True, 'type':'vk'}) else: social = doc logging.info(doc) logging.info(social) images = await get_images(db_name, file_ids, sign_avatar) async with httpx.AsyncClient(http2=True, verify=ctx) as client: attachments = [] for _id, img in images.items(): vk_photos_key = f"{item.get('_id')}:{_id}:{social.get('_id')}" doc_photo = await mongo_client['vk']['photos'].find_one({'id':vk_photos_key}) if doc_photo: attachments.append(doc_photo['attachment']) continue payload = {"v": "5.199"} if 'group' in social: payload['group_id'] = social['id'] resp = await client.post(f"{VK_API_URI}photos.getWallUploadServer", headers={"Authorization": f"Bearer {await token_manager.get_access_token(social['user_id'])}"}, data=payload, timeout=30.0) resp.raise_for_status() upload_url = resp.json()['response']['upload_url'] resp = await client.post(upload_url, files=[img], timeout=60.0) resp.raise_for_status() payload.update(resp.json()) payload.update({'caption':text}) resp = await client.post(f"{VK_API_URI}photos.saveWallPhoto", headers={"Authorization": f"Bearer {await token_manager.get_access_token(social['user_id'])}"}, data=payload, timeout=30.0) resp.raise_for_status() attachment = ','.join([f'photo{i['owner_id']}_{i['id']}' for i in resp.json()['response']]) if not attachment: return {'ok': False} await mongo_client['vk']['photos'].insert_one({'attachment':attachment,'id':vk_photos_key}) attachments.append(attachment) logging.info(attachments) resp = await client.post(f"{VK_API_URI}wall.edit", data={ "access_token": await token_manager.get_access_token(doc['user_id']), "owner_id": f'-{doc['id']}' if 'group' in doc else doc['id'], "post_id": post_id, 'message': text, 'attachments': attachments, 'primary_attachments_mode': 'grid', "v": "5.199", }) res = resp.json() res.update({'doc': doc}) if 'error' in res: return {'ok': False} update_spec = {"$push": {"vk": res}} await mongo_client[db_name][item.get('col')].update_one({'_id': ObjectId(item.get('_id'))}, update_spec) return {"ok": True} @router.post("/delete") async def wall_delete(data: dict, request: Request): # ---- auth ---- auth = request.headers.get("Authorization") if auth and auth.lower().startswith("bearer "): token = auth.split(" ", 1)[1].strip() if token != ATOKEN: return {"ok": False, "error": "unauthorized"} item = data["item"] post_id = data["post_id"] doc = data["doc"] logging.info(doc) owner_id = f"-{doc.get('id')}" if "group" in doc else doc.get("id") # ---- VK request ---- async with httpx.AsyncClient(http2=True, verify=ctx) as client: resp = await client.post(f"{VK_API_URI}wall.delete", data={ "access_token": await token_manager.get_access_token(doc["user_id"]), "owner_id": owner_id, "post_id": post_id, "v": "5.199", }) res = resp.json() logging.info(f"DELETE VK response: {res}") res["doc"] = doc if "error" in res: return {"ok": False, "error": res["error"]} # ---- Mongo remove vk post record ---- if item: col = item.get("col") mongo_col = mongo_client[item.get("db")][col] mongo_col.update_one( {"_id": ObjectId(item.get("_id"))}, {"$pull": {"vk": {"response": {"post_id": post_id}}}} ) return {"ok": True} @celery_app_vk.task( bind=True, rate_limit='60/m', name='vk.edit_vk', retry_backoff=True, retry_backoff_max=600, retry_jitter=True, max_retries=20, soft_time_limit=60, time_limit=100, ) def edit(self, post_id: int, item: dict, text: str, file_ids: list, social: dict, sign_avatar: bool = True): try: logging.info(f"VK начало edit {item} {file_ids}") res = httpx.post( 'http://127.0.0.1:30080/api/v1/social/vk/edit', headers={"Authorization": f"bearer {ATOKEN}"}, json={'post_id': post_id, 'item': item, 'text': text, 'file_ids': file_ids, 'doc': social, 'sign_avatar': sign_avatar}, timeout=100 ).json() if res['ok']: return res raise RuntimeError(f"VK send failed: {res}") except Exception as exc: raise self.retry(exc=exc) @celery_app_vk.task( bind=True, rate_limit='60/m', name='vk.wall_delete', retry_backoff=True, retry_backoff_max=600, retry_jitter=True, max_retries=20, soft_time_limit=60, time_limit=100, ) def delete_wall(self, post_id: int, doc: dict, item: dict|None = None): try: logging.info(f"VK удаления публикации {doc} {post_id}") res = httpx.post( 'http://127.0.0.1:30080/api/v1/social/vk/delete', headers={"Authorization": f"bearer {ATOKEN}"}, json={'item': item, 'post_id': post_id, "doc": doc}, timeout=100 ).json() if res.get('ok'): return res raise RuntimeError(f"VK send failed: {res}") except Exception as exc: raise self.retry(exc=exc) @celery_app_vk.task( bind=True, rate_limit='60/m', name='vk.send_vk', retry_backoff=True, retry_backoff_max=600, retry_jitter=True, max_retries=20, soft_time_limit=60, time_limit=100, ) def notify(self, item: dict, text: str, file_ids: list, social: dict, sign_avatar: bool = True): try: logging.info(f"VK начало публикации {item} {file_ids}") res = httpx.post( 'http://127.0.0.1:30080/api/v1/social/vk/send', headers={"Authorization": f"bearer {ATOKEN}"}, json={'item': item, 'text': text, 'file_ids': file_ids, 'doc': social, 'sign_avatar': sign_avatar}, timeout=100 ).json() if res.get('ok'): return res raise RuntimeError(f"VK send failed: {res}") except Exception as exc: raise self.retry(exc=exc)