332 lines
12 KiB
Python
332 lines
12 KiB
Python
# VK
|
|
|
|
## 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
|
|
from fastapi.responses import RedirectResponse
|
|
from images import insert_logo
|
|
|
|
### Конфигурация
|
|
|
|
celery_app_vk = Celery('vk ', broker=f'amqps://admin:{RABBITMQ_PASS}@{RABBITMQ_HOST}/vk', 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("/token")
|
|
async def oauth_callback(code: str, state: str, device_id: str = None):
|
|
token_data = await token_manager.get_token(code=code, state=state, device_id=device_id)
|
|
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"]
|
|
|
|
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)
|
|
|
|
owner_id = f"-{doc['id']}" if "group" in doc else doc["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 ----
|
|
col = item.get("col")
|
|
mongo_col = mongo_client[db_name][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://kultmost-api.kultmost/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, item: dict, post_id: int, doc: dict):
|
|
try:
|
|
logging.info(f"VK удаления публикации {item} {post_id}")
|
|
res = httpx.post(
|
|
'http://kultmost-api.kultmost/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://kultmost-api.kultmost/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)
|