389 lines
15 KiB
Python
389 lines
15 KiB
Python
# `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)
|