diff --git a/tasks.py b/tasks.py new file mode 100644 index 0000000..08f8956 --- /dev/null +++ b/tasks.py @@ -0,0 +1,721 @@ +from config import * +from db import * +from typing import Dict, Any, List, Optional, Callable, Union +import email_task as email_task +import vk as vk +import tg as tg +from utils import * +from models import NotifyItem +from auth import token_create +import asyncio + + +async def notify_in_profile(item: NotifyItem): + return await notify_collection.update_one( + { + "user_id": item.user_id, + "type": item.type_item, + "title": item.title, + "link": item.link, + "text": item.text, + }, + { + "$set": { + "user_id": item.user_id, + "type": item.type_item, + "title": item.title, + "text": item.text, + "link": item.link, + "read": True, + "prod": PROD, + } + }, + upsert=True, + ) + + +async def email_new_post(new_post: Dict): + async for user in users_collection.find({'roles':'partner'}): + login_link = f'{FULL_DOMAIN}/login/{await token_create(str(user['_id']), url=f'/paintings/{new_post['_id']}')}' + if PROD: + email_task.notify.delay(user['email'], { + 'text':'На сервисе опубликована новая картина', + 'list':[[ + {'label':'Название','text':new_post['title'], 'img':f"{FULL_DOMAIN}/b/posts/{new_post['medias_original'][0]}"}, + {'label':'Звание художника','text':new_post['grade']}, + {'label':'Город','text':new_post['city']}, + {'label':'Размер','text':new_post['size']}, + {'label':'Материал','text':new_post['material']}, + {'label':'Жанр','text':new_post['genre']}, + {'label':'Цена','text':new_post['price']}, + ]], + 'primary': {'href':login_link,'text':'Войти и посмотреть'}, + 'title': f"от художника {new_post['author']}", + 'login_link': login_link, + }, + 'Новая картина', 'main.html') + + +async def update_social_post(post): + pass + +async def publish_social_post(post): + query = { + 'active': True, + "$or": [{'service': True}, {'user_id': post['user']['_id']}] + } + post_ref = {'_id': post['_id'], 'db': 'posts', 'col': 'posts'} + medias = post.get("medias_original", []) + + platforms = [ + ('tg', tg, lambda s: [ + f"{FULL_PATH_DOMAIN}/a/{post['user']['_id']}" if should_sign_avatar(post, s) else None, + *[f"{FULL_PATH_DOMAIN}/b/posts/{m}" for m in medias] + ]), + ('vk', vk, lambda s: [ + post['user']['_id'] if should_sign_avatar(post, s) else None, + *medias + ]) + ] + + for platform_type, client, get_attachments in platforms: + if post.get(platform_type) != []: + continue + socials = [doc async for doc in social_collection.find({**query, 'type': platform_type})] + socials = {item["id"]: item for item in socials}.values() + for social in socials: + text = parse_posts_text(post, platform_type, social.get('service', False)) + sign_avatar = should_sign_avatar(post, social) + attachments = [a for a in get_attachments(social) if a is not None] + + client.notify.delay(post_ref, text, attachments, bson_clear(social), sign_avatar) + +async def user_social_competition(competition): + async for user in users_collection.find({'roles':'admin'}): + # email_task.notify.delay + if PROD: + pass + else: + print(competition) + +class SocialContentHandler: + """Base class for content type specific handlers""" + + def get_post_ref(self, item: Dict[str, Any]) -> Dict[str, Any]: + """Return reference for the content item""" + raise NotImplementedError + + def get_medias(self, item: Dict[str, Any]) -> List[Optional[str]]: + """Return list of media file_ids""" + raise NotImplementedError + + def get_text(self, item: Dict[str, Any]) -> str: + """Parse text for the content""" + raise NotImplementedError + + +class CompetitionHandler(SocialContentHandler): + def get_post_ref(self, item: Dict[str, Any]) -> Dict[str, Any]: + return {'_id': item['_id'], 'db': 'competitions', 'col': 'competitions'} + + def get_medias(self, item: Dict[str, Any]) -> List[Optional[str]]: + return [item.get("poster_file_id")] + + def get_text(self, item: Dict[str, Any]) -> str: + return parse_text_competition(item) + + +class PostHandler(SocialContentHandler): + def get_post_ref(self, item: Dict[str, Any]) -> Dict[str, Any]: + return {'_id': item['_id'], 'db': 'posts', 'col': 'posts'} + + def get_medias(self, item: Dict[str, Any]) -> List[Optional[str]]: + # Adjust according to your Post schema + return item.get("media_file_ids", []) or [item.get("main_image")] + + def get_text(self, item: Dict[str, Any]) -> str: + return item.get("text", "") + + +class NewsHandler(SocialContentHandler): + def get_post_ref(self, item: Dict[str, Any]) -> Dict[str, Any]: + return {'_id': item['_id'], 'db': 'news', 'col': 'news'} + + def get_medias(self, item: Dict[str, Any]) -> List[Optional[str]]: + return item.get("media_ids", []) + + def get_text(self, item: Dict[str, Any]) -> str: + return item.get("title", "") + "\n\n" + item.get("content", "") + + +# Registry for handlers +CONTENT_HANDLERS: Dict[str, SocialContentHandler] = { + "competition": CompetitionHandler(), + "post": PostHandler(), + "news": NewsHandler(), + # Add more: "user": UserProfileHandler(), etc. +} + + +class SocialPlatform: + """Platform configuration""" + def __init__( + self, + type_name: str, + client: Any, + get_attachments: Callable[[Dict], List[str]], + sign_avatar_func: Optional[Callable] = None + ): + self.type = type_name + self.client = client + self.get_attachments = get_attachments + self.sign_avatar_func = sign_avatar_func or should_sign_avatar + + +# Platform registry - easy to extend +PLATFORMS: Dict[str, SocialPlatform] = { + "tg": SocialPlatform( + type_name="tg", + client=tg, + get_attachments=lambda social, medias: [ + f"{FULL_PATH_DOMAIN}/a/{social.get('user_id')}" if should_sign_avatar(social) else None, + *[f"{FULL_PATH_DOMAIN}/b/competition/{m}" for m in medias if m] # generalize path + ] + ), + "vk": SocialPlatform( + type_name="vk", + client=vk, + get_attachments=lambda social, medias: [ + social.get('user_id') if should_sign_avatar(social) else None, + *medias + ] + ), + # Add more platforms easily: + # "ok": SocialPlatform(...), + # "telegram_channel": ... +} + + +async def get_socials_for_content( + content_type: str, + item: Dict[str, Any], + service_social: bool = False, + user_social: bool = False, + users_social: bool = False +) -> List[Dict]: + """Get relevant social accounts based on rules""" + query: Dict = {'active': True, "$or": []} + + if service_social: + query["$or"].append({'service': True}) + if user_social: + query["$or"].append({'user_id': item.get('user', {}).get('_id')}) + if users_social: + query["$or"].extend([ + {"competition": True}, + {"competition": {"$exists": False}} + ]) + + # Filter by platform types that exist in registry + platform_types = list(PLATFORMS.keys()) + cursor = social_collection.find({**query, 'type': {'$in': platform_types}}) + socials = [doc async for doc in cursor] + + # Deduplicate by id + return list({item["id"]: item for item in socials}.values()) + + +async def send_to_social( + content_type: str, + item: Dict[str, Any], + social: Dict[str, Any], + platform: SocialPlatform +) -> bool: + """Send one item to one social account with spam protection""" + handler = CONTENT_HANDLERS.get(content_type) + if not handler: + print(f"Unknown content type: {content_type}") + return False + + redis_key = f"nospamsends:{content_type}:{item['_id']}:{social['_id']}" + + already_sent = await redis_client.set( + redis_key, + "1", + ex=8 * 60 * 60, # 8 hours + nx=True + ) + + if not already_sent: + return False # already sent recently + + post_ref = handler.get_post_ref(item) + medias = handler.get_medias(item) + text = handler.get_text(item) + + sign_avatar = should_sign_avatar(item, social) if hasattr(handler, 'should_sign') else should_sign_avatar(item, social) + + attachments = [a for a in platform.get_attachments(social, medias) if a is not None] + + try: + platform.client.notify.delay( + post_ref, + text, + attachments, + bson_clear(social), + sign_avatar + ) + print(f"Queued {content_type} {item['_id']} to {platform.type} social {social['_id']}") + return True + except Exception as e: + print(f"Failed to queue to {platform.type}: {e}") + return False + + +async def service_social_publish( + content_type: str, + item: Dict[str, Any], + service_social: bool = False, + user_social: bool = False, + users_social: bool = False +): + """ + Generic social publishing service. + + Usage: + await service_social_publish("competition", competition_dict) + await service_social_publish("post", post_dict) + """ + if content_type not in CONTENT_HANDLERS: + raise ValueError(f"Unsupported content type: {content_type}") + + socials = await get_socials_for_content( + content_type, item, service_social, user_social, users_social + ) + + tasks = [] + for social in socials: + platform_key = social.get("type") + platform = PLATFORMS.get(platform_key) + if not platform: + continue + + tasks.append(send_to_social(content_type, item, social, platform)) + + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + + +# # Convenience wrappers +# async def service_social_competition(competition: Dict, **kwargs): +# return await service_social_publish("competition", competition, **kwargs) + + +# async def service_social_post(post: Dict, **kwargs): +# return await service_social_publish("post", post, **kwargs) + + +# async def service_social_news(news: Dict, **kwargs): +# return await service_social_publish("news", news, **kwargs) + +async def service_social_exhibition(exhibition): + query = { 'active': True, "$or": []} + query["$or"].append({'service': True}) + query["$or"].append({'user_id': exhibition['user']['_id']}) + + post_ref = {'_id': exhibition['_id'], 'db': 'exhibition', 'col': 'exhibition'} + medias = [exhibition.get("poster_orig_id")] + + platforms = [ + ('tg', tg, lambda s: [ + f"{FULL_PATH_DOMAIN}/a/{exhibition['user']['_id']}" if should_sign_avatar(exhibition, s) else None, + *[f"{FULL_PATH_DOMAIN}/b/exhibition/{m}" for m in medias] + ]), + ('vk', vk, lambda s: [ + exhibition['user']['_id'] if should_sign_avatar(exhibition, s) else None, + *medias + ]) + ] + + for platform_type, client, get_attachments in platforms: + if exhibition.get(platform_type) != []: + continue + socials = [doc async for doc in social_collection.find({**query, 'type': platform_type})] + socials = {item["id"]: item for item in socials}.values() + for social in socials: + text = parse_text_exhibition(exhibition) + sign_avatar = should_sign_avatar(exhibition, social) + attachments = [a for a in get_attachments(social) if a is not None] + redis_key = f"nospamsends:{exhibition['_id']}:{social['_id']}" + already_sent = await redis_client.set( + redis_key, + 1, + ex=8*60*60, + nx=True + ) + print(f"nospamsends:{exhibition['_id']}:{social['_id']}") + if already_sent: + print(social) + client.notify.delay(post_ref, text, attachments, bson_clear(social), sign_avatar) + +async def service_social_competition(competition, serviceSocial, userSocial, usersSocial): + query = { 'active': True, "$or": []} + if serviceSocial: + query["$or"].append({'service': True}) + + if userSocial: + query["$or"].append({'user_id': competition['user']['_id']}) + + if usersSocial: + query["$or"].append({"competition": True}) + query["$or"].append({"competition": {"$exists": False}}) + post_ref = {'_id': competition['_id'], 'db': 'competitions', 'col': 'competitions'} + medias = [competition.get("poster_file_id")] + + platforms = [ + ('tg', tg, lambda s: [ + f"{FULL_PATH_DOMAIN}/a/{competition['user']['_id']}" if should_sign_avatar(competition, s) else None, + *[f"{FULL_PATH_DOMAIN}/b/competition/{m}" for m in medias] + ]), + ('vk', vk, lambda s: [ + competition['user']['_id'] if should_sign_avatar(competition, s) else None, + *medias + ]) + ] + + for platform_type, client, get_attachments in platforms: + + socials = [doc async for doc in social_collection.find({**query, 'type': platform_type})] + socials = {item["id"]: item for item in socials}.values() + for social in socials: + text = parse_text_competition(competition) + sign_avatar = should_sign_avatar(competition, social) + attachments = [a for a in get_attachments(social) if a is not None] + redis_key = f"nospamsends:{competition['_id']}:{social['_id']}" + already_sent = await redis_client.set( + redis_key, + 1, + ex=8*60*60, + nx=True + ) + print(f"nospamsends:{competition['_id']}:{social['_id']}") + if already_sent: + print(social) + client.notify.delay(post_ref, text, attachments, bson_clear(social), sign_avatar) + +async def email_new_user(new_user): + async for user in users_collection.find({'roles':'admin'}): + login_link = f'{FULL_DOMAIN}/login/{await token_create(str(user['_id']), url=f'/artists/{new_user['_id']}')}' + role = {'user':'Художник','galerist':'Галерея',"reporter":'Репортер'}[new_user['roles'][0]] + if PROD: + email_task.notify.delay(user['email'], { + 'list':[[ + {'label':'ФИО','text': new_user['female_name_lastname'], 'img':f"{FULL_DOMAIN}/a/{new_user['_id']}"}, + {'label':'Звание','text': new_user['grade']} if new_user['roles'][0] != 'galerist' else {}, + {'label':'Город проживания','text': new_user['living_place']} + ]], + 'text': f'', + 'primary': {'href':login_link,'text':'Войти и посмотреть'}, + 'title': f"{role} {new_user['username']}", + 'login_link': login_link + }, + 'Новый пользователь', 'main.html') + +async def invite_users_to_competitions(doc): + new_notify = [] + + query = { + "roles": {"$in": ["user"]}, + "$or": [ + {"email_subscribed": True}, + {"email_subscribed": {"$exists": False}} + ] + } + + async for user in users_collection.find(query): + if not user.get("email"): + continue + + redis_key = f"nospam:{user['email']}:{doc['_id']}" + + already_sent = await redis_client.set( + redis_key, + 1, + ex=24*60*60, + nx=True + ) + + if already_sent: + new_notify.append( + notify_in_profile(NotifyItem( + user_id=str(user['_id']), + type_item='competition', + title='Приглашение на конкурс', + link=f'/profile/competition/{doc.get("id") or doc.get("_id")}', + text=f'«{doc['title']}» Организатор {doc['user']['username']}' + )) + ) + if PROD: + email_task.notify.delay(user['email'], {'competition': bson_clear(doc), 'login_link': f'{FULL_DOMAIN}/login/{await token_create(str(user['_id']), url=f'/competition/{doc.get("id") or doc.get("_id")}')}'}, 'Конкурс', 'competition.html') + await asyncio.gather(*new_notify) + +async def accept_competition_req(competition, user_id): + user = await users_collection.find_one({'_id': ObjectId(user_id)},{'email':1}) + login_link = f'{FULL_DOMAIN}/login/{await token_create(str(user['_id']), url=f'/profile/invitation')}' + if PROD: + email_task.notify.delay(user['email'], { + 'text':f'на конкурсе {competition['title']}', + 'list':[[ ]], + 'primary': {'href':login_link,'text':'Войти и посмотреть'}, + 'title': f"Вы прошли отбор", + 'login_link': login_link, + }, + 'Приглашение', 'main.html') + await notify_in_profile(NotifyItem(user_id=user_id, type_item='invation', title='Приглашение', link='/profile/invitation', text=f'Вы прошли отбор на конкурсе {competition['title']}')) + + +async def get_sender(post_id): + delivery_cursor = delivery_collection.find( + { + 'post_id': post_id,"prod": PROD, + 'status': { '$nin': ["Ожидает отправки", "Ожидает принятия условий", "Третье"] } + }, + {'receiver': 1} + ).sort('_id', 1) # 1 = ascending, -1 = descending (newest first) + + delivery = await delivery_cursor.limit(1).to_list(length=1) + delivery = delivery[0] if delivery else {} + return delivery.get('receiver') + +async def accept_invitations(invitations_id): + invitation = await invitations_collection.find_one_and_update( + {"_id": ObjectId(invitations_id)}, + {"$set": {"accept": True}}, + return_document=True, + ) + delivery_collection.update_many({'post_id': {'$in': [str(doc['_id']) for doc in invitation.get('posts', [])]}}, {'$set':{'status': 'Ожидает отправки'}}) + await notify_in_profile(NotifyItem( + user_id=invitation.get('exhibition',{}).get("user",{}).get('_id'), + type_item='c', + title=f'Выставка "{invitation.get('exhibition',{}).get("title")}"', + link=f'/profile/delivery', + text=f'Художник принял условия' + )) + +async def exhibition_send_invitations(exhibition_id): + exhibition = await exhibition_collection.find_one({'_id': ObjectId(exhibition_id)}) + exhibition_owner = exhibition.get('user') + obj_ids = [ObjectId(x) for x in exhibition.get('posts', [])] + posts = await posts_collection.find({"_id": {"$in": obj_ids}}).to_list() if obj_ids else [] + new_posts = {} + all_posts = {} + for post in posts: + post_id = str(post.get('_id')) + user = display_user(post.get('user')) + doc = await delivery_collection.update_one( + { + "exhibition_id": str(exhibition.get('_id')),"post_id": post_id, "status": "Ожидает принятия условий", "type": "exhibition", + },{ + "$set":{ + "sender": user, + "receiver": display_user(exhibition_owner), + "exhibition_id": str(exhibition.get('_id')), + "address": exhibition.get('address'), + "post_id": post_id, + "prod": PROD, + "img": post.get('medias',[0])[0], + "title": post.get('title'), + "status": "Ожидает принятия условий", + "type": "exhibition", + "updated_at": datetime.utcnow() + } + }, upsert=True) + if doc.upserted_id: + if (user_id:=user.get('_id')) in new_posts: + new_posts[user_id].append(post) + else: + new_posts.update({user_id: [post]}) + if (user_id:=user.get('_id')) in all_posts: + all_posts[user_id].append(post) + else: + all_posts.update({user_id: [post]}) + print(all_posts) + if all_posts: + for user_id, posts in all_posts.items(): + doc = await invitations_collection.update_one( + {'exhibition_id': str(exhibition.get('_id')),'user_id': user_id,'type': 'Выставка'}, + + {"$set":{'exhibition': bson_clear(exhibition), + 'exhibition_id': str(exhibition.get('_id')), + 'accept': False, + 'user':display_user(exhibition_owner), + 'user_id': user_id, + 'posts': posts, + 'prod': PROD, + 'type': 'Выставка', + 'item': exhibition.get('title')}}, + upsert=True + ) + if new_posts: + for user_id, posts in new_posts.items(): + await notify_in_profile(NotifyItem( + user_id=user_id, + type_item='invitation', + title='Приглашение', + link=f'/profile/invitation', + text=f'Приглашаем на выставку "{exhibition.get("title")}"' + )) + + user = posts[0].get('user') + token = await token_create( + str(user['_id']), + url=f'/profile/invitation#main' + ) + login_link = f'{FULL_DOMAIN}/login/{token}' + email_data = { + 'text': 'Приглашаем на выставку со следующими картинами: ', + 'list': [ + [ + {'label': 'Название', 'text': item.get('title', ''), 'img':f"{FULL_DOMAIN}/b/posts/{item['medias_original'][0]}"}, + ] + for item in posts + ], + 'primary': { + 'href': login_link, + 'text': 'Войти и посмотреть' + }, + 'title': exhibition.get('title', 'Выставка'), + 'login_link': login_link, + } + if PROD: + email_task.notify.delay( + user['email'], + email_data, + f'Приглашение на выставку {exhibition.title}', + 'main.html' + ) + + + +async def update_competition_req(competition, user, posts): + competition_id = competition.get('_id') + + if not user or not posts or not competition_id: + return + + posts_sorted = sorted(posts, key=lambda p: p.get('status') != 'Проходит') + + # Генерация login link + token = await token_create( + str(user['_id']), + url=f'/profile/delivery#main' + ) + login_link = f'{FULL_DOMAIN}/login/{token}' + + if True: + competition_owner = competition.get('user', {}) + + # === Создание доставок === + + post_ids_to_mark = [] # для последующего обновления delivery: True + post_ids_to_remove_delivery = [] + for post in posts: + post_id = str(post.get('_id')) + if post.get('status') == 'Проходит': + sender = await get_sender(post_id) or display_user(user) + delivery_collection.update_one({ + "competition_id": competition_id,"post_id": post_id, "status": "Ожидает отправки", + "sender._id": sender.get('_id'), "receiver._id": competition_owner.get('_id') + },{ + "$set":{ + "sender": sender, + "receiver": display_user(competition_owner), + "competition_id": competition_id, + "address": competition.get('address'), + "post_id": post_id, + "prod": PROD, + "img": post.get('medias',[0])[0], + "title": post.get('title'), + "status": "Ожидает отправки", + "type": "competition_reward", + "updated_at": datetime.utcnow() + } + }, upsert=True) + post_ids_to_mark.append(post_id) + elif post.get('status') != 'Проходит': + post_ids_to_remove_delivery.append(post_id) + + + if post_ids_to_remove_delivery: + await delivery_collection.delete_many({ + "competition_id": competition_id, + "status": "Ожидает отправки", + "post_id": {"$in": post_ids_to_remove_delivery} + }) + + await notify_in_profile(NotifyItem( + user_id=str(user['_id']), + type_item='invitation', + title='Приглашение', + link=f'/profile/delivery', + text=f'Вы прошли отбор на конкурсе "{competition.get("title")}"' + )) + all_allowed = all( + post.get('status') in ('Проходит', 'На рассмотрении') + for post in posts + ) + has_approved = any(p.get('status') == 'Проходит' for p in posts) + + # Email-уведомление + if user.get('email') and all_allowed and has_approved: + print([i['title'] for i in posts_sorted]) + if PROD and user.get('email') and all_allowed and has_approved: + email_data = { + 'text': 'Статус заявки обновлен', + 'list': [ + [ + {'label': 'Название', 'text': item.get('title', ''), 'img':f"{FULL_DOMAIN}/b/posts/{item['medias_original'][0]}"}, + {'label': 'Статус', 'text': item.get('status', '')} + ] + for item in posts_sorted + ], + 'primary': { + 'href': login_link, + 'text': 'Войти и посмотреть' + }, + 'title': competition.get('title', 'Конкурс'), + 'login_link': login_link, + } + + email_task.notify.delay( + user['email'], + email_data, + 'Обновление статуса заявки на конкурс', + 'main.html' + ) + +async def new_competiton_req(request): + competition = await competitions_col.find_one({'_id':ObjectId(request['competition_id'])}) + user = competition['user'] + if competition.get('type_') == 'artist': + link = f'/profile/partner/competition/{request['competition_id']}/{request['user']['_id']}' + else: + link = f'/profile/partner/competition/{request['competition_id']}' + login_link = f'{FULL_DOMAIN}/login/{await token_create(str(user['_id']), url=link)}' + print(user) + await notify_in_profile(NotifyItem( + user_id=user['_id'], + type_item='competition_req', + title='Новая заявка на конкурс', + link=link, + text=f'«{competition['title']}» От {request['user']['username']}' + )) + if PROD: + email_task.notify.delay(user['email'], { + 'list':[[{'label':'Название','text':i['title'], 'img':f"{FULL_DOMAIN}/b/posts/{i['medias'][0]}"},{'label':'Размер','text':i['size']},{'label':'Описание','text':i['description']}] for i in request['posts']], + 'text': f'Картины на конкурс «{competition['title']}»:', + 'primary': {'href':login_link,'text':'Войти и посмотреть'}, + 'title': f"От {request['user']['username']}", + 'login_link': login_link + }, + 'Новая заявка на конкурс', 'main.html') \ No newline at end of file