discours.io/core
5
0
Fork 0
Code Pull Requests Actions Projects Activity

Compare commits

base: discours.io:fix/sv-authorSortingBy
Branches Tags
discours.io:dev discours.io:feature/auth-internal discours.io:feat/sv-search-v4 discours.io:fix/sv-authorSortingBy discours.io:staging discours.io:main discours.io:feature/branched-pagination discours.io:feature/caching-improved
..
compare: discours.io:03c2d286ded482bf240c77d6730855e1c6d61064
Branches Tags
discours.io:dev discours.io:feature/auth-internal discours.io:feat/sv-search-v4 discours.io:fix/sv-authorSortingBy discours.io:staging discours.io:main discours.io:feature/branched-pagination discours.io:feature/caching-improved

1 Commits
fix/sv-aut ... 03c2d286de

Author SHA1 Message Date
Stepan Vladovskiy
03c2d286de fix: commented out Media in type graph ql shema for startup the core
Some checks failed
Deploy on push / deploy (push) Failing after 9s
Details
2025-01-25 16:58:53 -03:00
66 changed files with 1015 additions and 5191 deletions
Expand all files Collapse all files

5
.gitignore vendored
View File

@@ -159,7 +159,4 @@ dokku_config
views.json
*.pem
*.key
*.crt
*cache.json
.cursor
.devcontainer/devcontainer.json
*.crt

154
CHANGELOG.md
View File

@@ -1,135 +1,3 @@
#### [0.4.20] - 2025-05-03
- Исправлена ошибка в классе `CacheRevalidationManager`: добавлена инициализация атрибута `_redis`
- Улучшена обработка соединения с Redis в менеджере ревалидации кэша:
- Автоматическое восстановление соединения в случае его потери
- Проверка соединения перед выполнением операций с кэшем
- Дополнительное логирование для упрощения диагностики проблем
- Исправлен резолвер `unpublish_shout`:
- Корректное формирование синтетического поля `publication` с `published_at: null`
- Возвращение полноценного словаря с данными вместо объекта модели
- Улучшена загрузка связанных данных (авторы, темы) для правильного формирования ответа
#### [0.4.19] - 2025-04-14
- dropped `Shout.description` and `Draft.description` to be UX-generated
- use redis to init views counters after migrator
#### [0.4.18] - 2025-04-10
- Fixed `Topic.stat.authors` and `Topic.stat.comments`
- Fixed unique constraint violation for empty slug values:
- Modified `update_draft` resolver to handle empty slug values
- Modified `create_draft` resolver to prevent empty slug values
- Added validation to prevent inserting or updating drafts with empty slug
- Fixed database error "duplicate key value violates unique constraint draft_slug_key"
#### [0.4.17] - 2025-03-26
- Fixed `'Reaction' object is not subscriptable` error in hierarchical comments:
- Modified `get_reactions_with_stat()` to convert Reaction objects to dictionaries
- Added default values for limit/offset parameters
- Fixed `load_first_replies()` implementation with proper parameter passing
- Added doctest with example usage
- Limited child comments to 100 per parent for performance
#### [0.4.16] - 2025-03-22
- Added hierarchical comments pagination:
- Created new GraphQL query `load_comments_branch` for efficient loading of hierarchical comments
- Ability to load root comments with their first N replies
- Added pagination for both root and child comments
- Using existing `comments_count` field in `Stat` type to display number of replies
- Added special `first_replies` field to store first replies to a comment
- Optimized SQL queries for efficient loading of comment hierarchies
- Implemented flexible comment sorting system (by time, rating)
#### [0.4.15] - 2025-03-22
- Upgraded caching system described `docs/caching.md`
- Module `cache/memorycache.py` removed
- Enhanced caching system with backward compatibility:
- Unified cache key generation with support for existing naming patterns
- Improved Redis operation function with better error handling
- Updated precache module to use consistent Redis interface
- Integrated revalidator with the invalidation system for better performance
- Added comprehensive documentation for the caching system
- Enhanced cached_query to support template-based cache keys
- Standardized error handling across all cache operations
- Optimized cache invalidation system:
- Added targeted invalidation for individual entities (authors, topics)
- Improved revalidation manager with individual object processing
- Implemented batched processing for high-volume invalidations
- Reduced Redis operations by using precise key invalidation instead of prefix-based wipes
- Added special handling for slug changes in topics
- Unified caching system for all models:
- Implemented abstract functions `cache_data`, `get_cached_data` and `invalidate_cache_by_prefix`
- Added `cached_query` function for unified approach to query caching
- Updated resolvers `author.py` and `topic.py` to use the new caching API
- Improved logging for cache operations to simplify debugging
- Optimized Redis memory usage through key format unification
- Improved caching and sorting in Topic and Author modules:
- Added support for dictionary sorting parameters in `by` for both modules
- Optimized cache key generation for stable behavior with various parameters
- Enhanced sorting logic with direction support and arbitrary fields
- Added `by` parameter support in the API for getting topics by community
- Performance optimizations for author-related queries:
- Added SQLAlchemy-managed indexes to `Author`, `AuthorFollower`, `AuthorRating` and `AuthorBookmark` models
- Implemented persistent Redis caching for author queries without TTL (invalidated only on changes)
- Optimized author retrieval with separate endpoints:
- `get_authors_all` - returns all non-deleted authors without statistics
- `load_authors_by` - optimized to use caching and efficient sorting and pagination
- Improved SQL queries with optimized JOIN conditions and efficient filtering
- Added pre-aggregation of statistics (shouts count, followers count) in single efficient queries
- Implemented robust cache invalidation on author updates
- Created necessary indexes for author lookups by user ID, slug, and timestamps
#### [0.4.14] - 2025-03-21
- Significant performance improvements for topic queries:
- Added database indexes to optimize JOIN operations
- Implemented persistent Redis caching for topic queries (no TTL, invalidated only on changes)
- Optimized topic retrieval with separate endpoints for different use cases:
- `get_topics_all` - returns all topics without statistics for lightweight listing
- `get_topics_by_community` - adds pagination and optimized filtering by community
- Added SQLAlchemy-managed indexes directly in ORM models for automatic schema maintenance
- Created `sync_indexes()` function for automatic index synchronization during app startup
- Reduced database load by pre-aggregating statistics in optimized SQL queries
- Added robust cache invalidation on topic create/update/delete operations
- Improved query optimization with proper JOIN conditions and specific partial indexes
#### [0.4.13] - 2025-03-20
- Fixed Topic objects serialization error in cache/memorycache.py
- Improved CustomJSONEncoder to support SQLAlchemy models with dict() method
- Enhanced error handling in cache_on_arguments decorator
- Modified `load_reactions_by` to include deleted reactions when `include_deleted=true` for proper comment tree building
- Fixed featured/unfeatured logic in reaction processing:
- Dislike reactions now properly take precedence over likes
- Featured status now requires more than 4 likes from users with featured articles
- Removed unnecessary filters for deleted reactions since rating reactions are physically deleted
- Author's featured status now based on having non-deleted articles with featured_at
#### [0.4.12] - 2025-03-19
- `delete_reaction` detects comments and uses `deleted_at` update
- `check_to_unfeature` etc. update
- dogpile dep in `services/memorycache.py` optimized
#### [0.4.11] - 2025-02-12
- `create_draft` resolver requires draft_id fixed
- `create_draft` resolver defaults body and title fields to empty string
#### [0.4.9] - 2025-02-09
- `Shout.draft` field added
- `Draft` entity added
- `create_draft`, `update_draft`, `delete_draft` mutations and resolvers added
- `create_shout`, `update_shout`, `delete_shout` mutations removed from GraphQL API
- `load_drafts` resolver implemented
- `publish_` and `unpublish_` mutations and resolvers added
- `create_`, `update_`, `delete_` mutations and resolvers added for `Draft` entity
- tests with pytest for original auth, shouts, drafts
- `Dockerfile` and `pyproject.toml` removed for the simplicity: `Procfile` and `requirements.txt`
#### [0.4.8] - 2025-02-03
- `Reaction.deleted_at` filter on `update_reaction` resolver added
- `triggers` module updated with `after_shout_handler`, `after_reaction_handler` for cache revalidation
- `after_shout_handler`, `after_reaction_handler` now also handle `deleted_at` field
- `get_cached_topic_followers` fixed
- `get_my_rates_comments` fixed
#### [0.4.7]
- `get_my_rates_shouts` resolver added with:
- `shout_id` and `my_rate` fields in response
@@ -181,7 +49,7 @@
#### [0.4.4]
- `followers_stat` removed for shout
- sqlite3 support added
- `rating_stat` and `comments_count` fixes
- `rating_stat` and `commented_stat` fixes
#### [0.4.3]
- cache reimplemented
@@ -337,4 +205,22 @@
#### [0.2.7]
- `loadFollowedReactions` now with `
- `loadFollowedReactions` now with `login_required`
- notifier service api draft
- added `shout` visibility kind in schema
- community isolated from author in orm
#### [0.2.6]
- redis connection pool
- auth context fixes
- communities orm, resolvers, schema
#### [0.2.5]
- restructured
- all users have their profiles as authors in core
- `gittask`, `inbox` and `auth` logics removed
- `settings` moved to base and now smaller
- new outside auth schema
- removed `gittask`, `auth`, `inbox`, `migration`

25
Dockerfile
View File

@@ -1,18 +1,25 @@
FROM python:slim
FROM python:3.12-alpine
RUN apt-get update && apt-get install -y \
postgresql-client \
curl \
&& rm -rf /var/lib/apt/lists/*
# Update package lists and install necessary dependencies
RUN apk update && \
apk add --no-cache build-base icu-data-full curl python3-dev musl-dev && \
curl -sSL https://install.python-poetry.org | python
# Set working directory
WORKDIR /app
COPY requirements.txt .
# Copy only the pyproject.toml file initially
COPY pyproject.toml /app/
RUN pip install -r requirements.txt
# Install poetry and dependencies
RUN pip install poetry && \
poetry config virtualenvs.create false && \
poetry install --no-root --only main
COPY . .
# Copy the rest of the files
COPY . /app
# Expose the port
EXPOSE 8000
CMD ["python", "-m", "granian", "main:app", "--interface", "asgi", "--host", "0.0.0.0", "--port", "8000"]
CMD ["python", "server.py"]

1
Procfile Normal file
View File

@@ -0,0 +1 @@
web: python server.py

31
README.md
View File

@@ -37,43 +37,36 @@ Backend service providing GraphQL API for content management system with reactio
## Development
### Prepare environment:
### Setup
Start API server with `dev` keyword added and `mkcert` installed:
```shell
mkdir .venv
python3.12 -m venv venv
source venv/bin/activate
```
python3.12 -m venv .venv
poetry env use .venv/bin/python3.12
poetry update
### Run server
First, certifcates are required to run the server.
```shell
mkcert -install
mkcert localhost
```
Then, run the server:
```shell
python -m granian main:app --interface asgi --host 0.0.0.0 --port 8000
poetry run server.py dev
```
### Useful Commands
```shell
# Linting and import sorting
ruff check . --fix --select I
poetry run ruff check . --fix --select I
# Code formatting
ruff format . --line-length=120
poetry run ruff format . --line-length=120
# Run tests
pytest
poetry run pytest
# Type checking
mypy .
poetry run mypy .
```
### Code Style

13
auth/identity.py
View File

@@ -1,15 +1,14 @@
from binascii import hexlify
from hashlib import sha256
# from base.exceptions import InvalidPassword, InvalidToken
from base.orm import local_session
from jwt import DecodeError, ExpiredSignatureError
from passlib.hash import bcrypt
from auth.exceptions import ExpiredToken, InvalidToken
from auth.jwtcodec import JWTCodec
from auth.tokenstorage import TokenStorage
from orm.user import User
# from base.exceptions import InvalidPassword, InvalidToken
from services.db import local_session
from orm import User
class Password:
@@ -80,10 +79,10 @@ class Identity:
if not await TokenStorage.exist(f"{payload.user_id}-{payload.username}-{token}"):
# raise InvalidToken("Login token has expired, please login again")
return {"error": "Token has expired"}
except ExpiredToken:
except ExpiredSignatureError:
# raise InvalidToken("Login token has expired, please try again")
return {"error": "Token has expired"}
except InvalidToken:
except DecodeError:
# raise InvalidToken("token format error") from e
return {"error": "Token format error"}
with local_session() as session:

5
auth/tokenstorage.py
View File

@@ -1,8 +1,9 @@
from datetime import datetime, timedelta, timezone
from base.redis import redis
from validations.auth import AuthInput
from auth.jwtcodec import JWTCodec
from auth.validations import AuthInput
from services.redis import redis
from settings import ONETIME_TOKEN_LIFE_SPAN, SESSION_TOKEN_LIFE_SPAN

110
auth/usermodel.py Normal file
View File

@@ -0,0 +1,110 @@
import time
from sqlalchemy import JSON, Boolean, Column, DateTime, ForeignKey, Integer, String, func
from sqlalchemy.orm import relationship
from services.db import Base
class Permission(Base):
__tablename__ = "permission"
id = Column(String, primary_key=True, unique=True, nullable=False, default=None)
resource = Column(String, nullable=False)
operation = Column(String, nullable=False)
class Role(Base):
__tablename__ = "role"
id = Column(String, primary_key=True, unique=True, nullable=False, default=None)
name = Column(String, nullable=False)
permissions = relationship(Permission)
class AuthorizerUser(Base):
__tablename__ = "authorizer_users"
id = Column(String, primary_key=True, unique=True, nullable=False, default=None)
key = Column(String)
email = Column(String, unique=True)
email_verified_at = Column(Integer)
family_name = Column(String)
gender = Column(String)
given_name = Column(String)
is_multi_factor_auth_enabled = Column(Boolean)
middle_name = Column(String)
nickname = Column(String)
password = Column(String)
phone_number = Column(String, unique=True)
phone_number_verified_at = Column(Integer)
# preferred_username = Column(String, nullable=False)
picture = Column(String)
revoked_timestamp = Column(Integer)
roles = Column(String, default="author,reader")
signup_methods = Column(String, default="magic_link_login")
created_at = Column(Integer, default=lambda: int(time.time()))
updated_at = Column(Integer, default=lambda: int(time.time()))
class UserRating(Base):
__tablename__ = "user_rating"
id = None
rater: Column = Column(ForeignKey("user.id"), primary_key=True, index=True)
user: Column = Column(ForeignKey("user.id"), primary_key=True, index=True)
value: Column = Column(Integer)
@staticmethod
def init_table():
pass
class UserRole(Base):
__tablename__ = "user_role"
id = None
user = Column(ForeignKey("user.id"), primary_key=True, index=True)
role = Column(ForeignKey("role.id"), primary_key=True, index=True)
class User(Base):
__tablename__ = "user"
default_user = None
email = Column(String, unique=True, nullable=False, comment="Email")
username = Column(String, nullable=False, comment="Login")
password = Column(String, nullable=True, comment="Password")
bio = Column(String, nullable=True, comment="Bio") # status description
about = Column(String, nullable=True, comment="About") # long and formatted
userpic = Column(String, nullable=True, comment="Userpic")
name = Column(String, nullable=True, comment="Display name")
slug = Column(String, unique=True, comment="User's slug")
links = Column(JSON, nullable=True, comment="Links")
oauth = Column(String, nullable=True)
oid = Column(String, nullable=True)
muted = Column(Boolean, default=False)
confirmed = Column(Boolean, default=False)
created_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now(), comment="Created at")
updated_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now(), comment="Updated at")
last_seen = Column(DateTime(timezone=True), nullable=False, server_default=func.now(), comment="Was online at")
deleted_at = Column(DateTime(timezone=True), nullable=True, comment="Deleted at")
ratings = relationship(UserRating, foreign_keys=UserRating.user)
roles = relationship(lambda: Role, secondary=UserRole.__tablename__)
def get_permission(self):
scope = {}
for role in self.roles:
for p in role.permissions:
if p.resource not in scope:
scope[p.resource] = set()
scope[p.resource].add(p.operation)
print(scope)
return scope
# if __name__ == "__main__":
# print(User.get_permission(user_id=1))

116
auth/validations.py
View File

@@ -1,116 +0,0 @@
import re
from datetime import datetime
from typing import Dict, List, Optional, Union
from pydantic import BaseModel, Field, field_validator
# RFC 5322 compliant email regex pattern
EMAIL_PATTERN = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
class AuthInput(BaseModel):
"""Base model for authentication input validation"""
user_id: str = Field(description="Unique user identifier")
username: str = Field(min_length=2, max_length=50)
token: str = Field(min_length=32)
@field_validator("user_id")
@classmethod
def validate_user_id(cls, v: str) -> str:
if not v.strip():
raise ValueError("user_id cannot be empty")
return v
class UserRegistrationInput(BaseModel):
"""Validation model for user registration"""
email: str = Field(max_length=254) # Max email length per RFC 5321
password: str = Field(min_length=8, max_length=100)
name: str = Field(min_length=2, max_length=50)
@field_validator("email")
@classmethod
def validate_email(cls, v: str) -> str:
"""Validate email format"""
if not re.match(EMAIL_PATTERN, v):
raise ValueError("Invalid email format")
return v.lower()
@field_validator("password")
@classmethod
def validate_password_strength(cls, v: str) -> str:
"""Validate password meets security requirements"""
if not any(c.isupper() for c in v):
raise ValueError("Password must contain at least one uppercase letter")
if not any(c.islower() for c in v):
raise ValueError("Password must contain at least one lowercase letter")
if not any(c.isdigit() for c in v):
raise ValueError("Password must contain at least one number")
if not any(c in "!@#$%^&*()_+-=[]{}|;:,.<>?" for c in v):
raise ValueError("Password must contain at least one special character")
return v
class UserLoginInput(BaseModel):
"""Validation model for user login"""
email: str = Field(max_length=254)
password: str = Field(min_length=8, max_length=100)
@field_validator("email")
@classmethod
def validate_email(cls, v: str) -> str:
if not re.match(EMAIL_PATTERN, v):
raise ValueError("Invalid email format")
return v.lower()
class TokenPayload(BaseModel):
"""Validation model for JWT token payload"""
user_id: str
username: str
exp: datetime
iat: datetime
scopes: Optional[List[str]] = []
class OAuthInput(BaseModel):
"""Validation model for OAuth input"""
provider: str = Field(pattern="^(google|github|facebook)$")
code: str
redirect_uri: Optional[str] = None
@field_validator("provider")
@classmethod
def validate_provider(cls, v: str) -> str:
valid_providers = ["google", "github", "facebook"]
if v.lower() not in valid_providers:
raise ValueError(f"Provider must be one of: {', '.join(valid_providers)}")
return v.lower()
class AuthResponse(BaseModel):
"""Validation model for authentication responses"""
success: bool
token: Optional[str] = None
error: Optional[str] = None
user: Optional[Dict[str, Union[str, int, bool]]] = None
@field_validator("error")
@classmethod
def validate_error_if_not_success(cls, v: Optional[str], info) -> Optional[str]:
if not info.data.get("success") and not v:
raise ValueError("Error message required when success is False")
return v
@field_validator("token")
@classmethod
def validate_token_if_success(cls, v: Optional[str], info) -> Optional[str]:
if info.data.get("success") and not v:
raise ValueError("Token required when success is True")
return v

415
cache/cache.py vendored
View File

@@ -1,37 +1,7 @@
"""
Caching system for the Discours platform
----------------------------------------
This module provides a comprehensive caching solution with these key components:
1. KEY NAMING CONVENTIONS:
- Entity-based keys: "entity:property:value" (e.g., "author:id:123")
- Collection keys: "entity:collection:params" (e.g., "authors:stats:limit=10:offset=0")
- Special case keys: Maintained for backwards compatibility (e.g., "topic_shouts_123")
2. CORE FUNCTIONS:
- cached_query(): High-level function for retrieving cached data or executing queries
3. ENTITY-SPECIFIC FUNCTIONS:
- cache_author(), cache_topic(): Cache entity data
- get_cached_author(), get_cached_topic(): Retrieve entity data from cache
- invalidate_cache_by_prefix(): Invalidate all keys with a specific prefix
4. CACHE INVALIDATION STRATEGY:
- Direct invalidation via invalidate_* functions for immediate changes
- Delayed invalidation via revalidation_manager for background processing
- Event-based triggers for automatic cache updates (see triggers.py)
To maintain consistency with the existing codebase, this module preserves
the original key naming patterns while providing a more structured approach
for new cache operations.
"""
import asyncio
import json
from typing import Any, Dict, List, Optional, Union
from typing import List
import orjson
from sqlalchemy import and_, join, select
from orm.author import Author, AuthorFollower
@@ -49,25 +19,11 @@ DEFAULT_FOLLOWS = {
"communities": [{"id": 1, "name": "Дискурс", "slug": "discours", "pic": ""}],
}
CACHE_TTL = 300 # 5 minutes
# Key templates for common entity types
# These are used throughout the codebase and should be maintained for compatibility
CACHE_KEYS = {
"TOPIC_ID": "topic:id:{}",
"TOPIC_SLUG": "topic:slug:{}",
"TOPIC_AUTHORS": "topic:authors:{}",
"TOPIC_FOLLOWERS": "topic:followers:{}",
"TOPIC_SHOUTS": "topic_shouts_{}",
"AUTHOR_ID": "author:id:{}",
"AUTHOR_USER": "author:user:{}",
"SHOUTS": "shouts:{}",
}
# Cache topic data
async def cache_topic(topic: dict):
payload = json.dumps(topic, cls=CustomJSONEncoder)
# Cache by id and slug for quick access
await asyncio.gather(
redis.execute("SET", f"topic:id:{topic['id']}", payload),
redis.execute("SET", f"topic:slug:{topic['slug']}", payload),
@@ -77,6 +33,7 @@ async def cache_topic(topic: dict):
# Cache author data
async def cache_author(author: dict):
payload = json.dumps(author, cls=CustomJSONEncoder)
# Cache author data by user and id
await asyncio.gather(
redis.execute("SET", f"author:user:{author['user'].strip()}", str(author["id"])),
redis.execute("SET", f"author:id:{author['id']}", payload),
@@ -86,22 +43,22 @@ async def cache_author(author: dict):
# Cache follows data
async def cache_follows(follower_id: int, entity_type: str, entity_id: int, is_insert=True):
key = f"author:follows-{entity_type}s:{follower_id}"
follows_str = await redis.execute("GET", key)
follows = orjson.loads(follows_str) if follows_str else DEFAULT_FOLLOWS[entity_type]
follows_str = await redis.execute("get", key)
follows = json.loads(follows_str) if follows_str else DEFAULT_FOLLOWS[entity_type]
if is_insert:
if entity_id not in follows:
follows.append(entity_id)
else:
follows = [eid for eid in follows if eid != entity_id]
await redis.execute("SET", key, json.dumps(follows, cls=CustomJSONEncoder))
await redis.execute("set", key, json.dumps(follows, cls=CustomJSONEncoder))
await update_follower_stat(follower_id, entity_type, len(follows))
# Update follower statistics
async def update_follower_stat(follower_id, entity_type, count):
follower_key = f"author:id:{follower_id}"
follower_str = await redis.execute("GET", follower_key)
follower = orjson.loads(follower_str) if follower_str else None
follower_str = await redis.execute("get", follower_key)
follower = json.loads(follower_str) if follower_str else None
if follower:
follower["stat"] = {f"{entity_type}s": count}
await cache_author(follower)
@@ -110,9 +67,9 @@ async def update_follower_stat(follower_id, entity_type, count):
# Get author from cache
async def get_cached_author(author_id: int, get_with_stat):
author_key = f"author:id:{author_id}"
result = await redis.execute("GET", author_key)
result = await redis.execute("get", author_key)
if result:
return orjson.loads(result)
return json.loads(result)
# Load from database if not found in cache
q = select(Author).where(Author.id == author_id)
authors = get_with_stat(q)
@@ -135,16 +92,16 @@ async def get_cached_topic(topic_id: int):
dict: Topic data or None if not found.
"""
topic_key = f"topic:id:{topic_id}"
cached_topic = await redis.execute("GET", topic_key)
cached_topic = await redis.execute("get", topic_key)
if cached_topic:
return orjson.loads(cached_topic)
return json.loads(cached_topic)
# If not in cache, fetch from the database
with local_session() as session:
topic = session.execute(select(Topic).where(Topic.id == topic_id)).scalar_one_or_none()
if topic:
topic_dict = topic.dict()
await redis.execute("SET", topic_key, json.dumps(topic_dict, cls=CustomJSONEncoder))
await redis.execute("set", topic_key, json.dumps(topic_dict, cls=CustomJSONEncoder))
return topic_dict
return None
@@ -153,9 +110,9 @@ async def get_cached_topic(topic_id: int):
# Get topic by slug from cache
async def get_cached_topic_by_slug(slug: str, get_with_stat):
topic_key = f"topic:slug:{slug}"
result = await redis.execute("GET", topic_key)
result = await redis.execute("get", topic_key)
if result:
return orjson.loads(result)
return json.loads(result)
# Load from database if not found in cache
topic_query = select(Topic).where(Topic.slug == slug)
topics = get_with_stat(topic_query)
@@ -170,8 +127,8 @@ async def get_cached_topic_by_slug(slug: str, get_with_stat):
async def get_cached_authors_by_ids(author_ids: List[int]) -> List[dict]:
# Fetch all author data concurrently
keys = [f"author:id:{author_id}" for author_id in author_ids]
results = await asyncio.gather(*(redis.execute("GET", key) for key in keys))
authors = [orjson.loads(result) if result else None for result in results]
results = await asyncio.gather(*(redis.execute("get", key) for key in keys))
authors = [json.loads(result) if result else None for result in results]
# Load missing authors from database and cache
missing_indices = [index for index, author in enumerate(authors) if author is None]
if missing_indices:
@@ -188,47 +145,47 @@ async def get_cached_authors_by_ids(author_ids: List[int]) -> List[dict]:
async def get_cached_topic_followers(topic_id: int):
"""
Получает подписчиков темы по ID, используя кеш Redis.
Если данные отсутствуют в кеше, извлекает из базы данных и кеширует их.
Args:
topic_id: ID темы
Returns:
List[dict]: Список подписчиков с их данными
:param topic_id: Идентификатор темы, подписчиков которой необходимо получить.
:return: Список подписчиков темы, каждый элемент представляет собой словарь с ID и именем автора.
"""
try:
cache_key = CACHE_KEYS["TOPIC_FOLLOWERS"].format(topic_id)
cached = await redis.execute("GET", cache_key)
# Попытка получить данные из кеша
cached = await redis.get(f"topic:followers:{topic_id}")
if cached:
followers_ids = orjson.loads(cached)
logger.debug(f"Found {len(followers_ids)} cached followers for topic #{topic_id}")
return await get_cached_authors_by_ids(followers_ids)
with local_session() as session:
followers_ids = [
f[0]
for f in session.query(Author.id)
.join(TopicFollower, TopicFollower.follower == Author.id)
.filter(TopicFollower.topic == topic_id)
.all()
]
await redis.execute("SETEX", cache_key, CACHE_TTL, orjson.dumps(followers_ids))
followers_ids = json.loads(cached)
logger.debug(f"Cached {len(followers_ids)} followers for topic #{topic_id}")
followers = await get_cached_authors_by_ids(followers_ids)
logger.debug(f"Cached {len(followers)} followers for topic #{topic_id}")
return followers
# Если данные не найдены в кеше, загрузка из базы данных
async with local_session() as session:
result = await session.execute(
session.query(Author.id)
.join(TopicFollower, TopicFollower.follower == Author.id)
.filter(TopicFollower.topic == topic_id)
)
followers_ids = [f[0] for f in result.scalars().all()]
# Кеширование результатов
await redis.set(f"topic:followers:{topic_id}", json.dumps(followers_ids))
# Получение подробной информации о подписчиках по их ID
followers = await get_cached_authors_by_ids(followers_ids)
logger.debug(followers)
return followers
except Exception as e:
logger.error(f"Error getting followers for topic #{topic_id}: {str(e)}")
logger.error(f"Ошибка при получении подписчиков для темы #{topic_id}: {str(e)}")
return []
# Get cached author followers
async def get_cached_author_followers(author_id: int):
# Check cache for data
cached = await redis.execute("GET", f"author:followers:{author_id}")
cached = await redis.execute("get", f"author:followers:{author_id}")
if cached:
followers_ids = orjson.loads(cached)
followers_ids = json.loads(cached)
followers = await get_cached_authors_by_ids(followers_ids)
logger.debug(f"Cached followers for author #{author_id}: {len(followers)}")
return followers
@@ -242,7 +199,7 @@ async def get_cached_author_followers(author_id: int):
.filter(AuthorFollower.author == author_id, Author.id != author_id)
.all()
]
await redis.execute("SET", f"author:followers:{author_id}", orjson.dumps(followers_ids))
await redis.execute("SET", f"author:followers:{author_id}", json.dumps(followers_ids))
followers = await get_cached_authors_by_ids(followers_ids)
return followers
@@ -250,9 +207,9 @@ async def get_cached_author_followers(author_id: int):
# Get cached follower authors
async def get_cached_follower_authors(author_id: int):
# Attempt to retrieve authors from cache
cached = await redis.execute("GET", f"author:follows-authors:{author_id}")
cached = await redis.execute("get", f"author:follows-authors:{author_id}")
if cached:
authors_ids = orjson.loads(cached)
authors_ids = json.loads(cached)
else:
# Query authors from database
with local_session() as session:
@@ -264,7 +221,7 @@ async def get_cached_follower_authors(author_id: int):
.where(AuthorFollower.follower == author_id)
).all()
]
await redis.execute("SET", f"author:follows-authors:{author_id}", orjson.dumps(authors_ids))
await redis.execute("SET", f"author:follows-authors:{author_id}", json.dumps(authors_ids))
authors = await get_cached_authors_by_ids(authors_ids)
return authors
@@ -273,9 +230,9 @@ async def get_cached_follower_authors(author_id: int):
# Get cached follower topics
async def get_cached_follower_topics(author_id: int):
# Attempt to retrieve topics from cache
cached = await redis.execute("GET", f"author:follows-topics:{author_id}")
cached = await redis.execute("get", f"author:follows-topics:{author_id}")
if cached:
topics_ids = orjson.loads(cached)
topics_ids = json.loads(cached)
else:
# Load topics from database and cache them
with local_session() as session:
@@ -286,13 +243,13 @@ async def get_cached_follower_topics(author_id: int):
.where(TopicFollower.follower == author_id)
.all()
]
await redis.execute("SET", f"author:follows-topics:{author_id}", orjson.dumps(topics_ids))
await redis.execute("SET", f"author:follows-topics:{author_id}", json.dumps(topics_ids))
topics = []
for topic_id in topics_ids:
topic_str = await redis.execute("GET", f"topic:id:{topic_id}")
topic_str = await redis.execute("get", f"topic:id:{topic_id}")
if topic_str:
topic = orjson.loads(topic_str)
topic = json.loads(topic_str)
if topic and topic not in topics:
topics.append(topic)
@@ -312,12 +269,12 @@ async def get_cached_author_by_user_id(user_id: str, get_with_stat):
dict: Dictionary with author data or None if not found.
"""
# Attempt to find author ID by user_id in Redis cache
author_id = await redis.execute("GET", f"author:user:{user_id.strip()}")
author_id = await redis.execute("get", f"author:user:{user_id.strip()}")
if author_id:
# If ID is found, get full author data by ID
author_data = await redis.execute("GET", f"author:id:{author_id}")
author_data = await redis.execute("get", f"author:id:{author_id}")
if author_data:
return orjson.loads(author_data)
return json.loads(author_data)
# If data is not found in cache, query the database
author_query = select(Author).where(Author.user == user_id)
@@ -328,7 +285,7 @@ async def get_cached_author_by_user_id(user_id: str, get_with_stat):
author_dict = author.dict()
await asyncio.gather(
redis.execute("SET", f"author:user:{user_id.strip()}", str(author.id)),
redis.execute("SET", f"author:id:{author.id}", orjson.dumps(author_dict)),
redis.execute("SET", f"author:id:{author.id}", json.dumps(author_dict)),
)
return author_dict
@@ -349,9 +306,9 @@ async def get_cached_topic_authors(topic_id: int):
"""
# Attempt to get a list of author IDs from cache
rkey = f"topic:authors:{topic_id}"
cached_authors_ids = await redis.execute("GET", rkey)
cached_authors_ids = await redis.execute("get", rkey)
if cached_authors_ids:
authors_ids = orjson.loads(cached_authors_ids)
authors_ids = json.loads(cached_authors_ids)
else:
# If cache is empty, get data from the database
with local_session() as session:
@@ -363,7 +320,7 @@ async def get_cached_topic_authors(topic_id: int):
)
authors_ids = [author_id for (author_id,) in session.execute(query).all()]
# Cache the retrieved author IDs
await redis.execute("SET", rkey, orjson.dumps(authors_ids))
await redis.execute("set", rkey, json.dumps(authors_ids))
# Retrieve full author details from cached IDs
if authors_ids:
@@ -372,257 +329,3 @@ async def get_cached_topic_authors(topic_id: int):
return authors
return []
async def invalidate_shouts_cache(cache_keys: List[str]):
"""
Инвалидирует кэш выборок публикаций по переданным ключам.
"""
for key in cache_keys:
try:
# Формируем полный ключ кэша
cache_key = f"shouts:{key}"
# Удаляем основной кэш
await redis.execute("DEL", cache_key)
logger.debug(f"Invalidated cache key: {cache_key}")
# Добавляем ключ в список инвалидированных с TTL
await redis.execute("SETEX", f"{cache_key}:invalidated", CACHE_TTL, "1")
# Если это кэш темы, инвалидируем также связанные ключи
if key.startswith("topic_"):
topic_id = key.split("_")[1]
related_keys = [
f"topic:id:{topic_id}",
f"topic:authors:{topic_id}",
f"topic:followers:{topic_id}",
f"topic:stats:{topic_id}",
]
for related_key in related_keys:
await redis.execute("DEL", related_key)
logger.debug(f"Invalidated related key: {related_key}")
except Exception as e:
logger.error(f"Error invalidating cache key {key}: {e}")
async def cache_topic_shouts(topic_id: int, shouts: List[dict]):
"""Кэширует список публикаций для темы"""
key = f"topic_shouts_{topic_id}"
payload = json.dumps(shouts, cls=CustomJSONEncoder)
await redis.execute("SETEX", key, CACHE_TTL, payload)
async def get_cached_topic_shouts(topic_id: int) -> List[dict]:
"""Получает кэшированный список публикаций для темы"""
key = f"topic_shouts_{topic_id}"
cached = await redis.execute("GET", key)
if cached:
return orjson.loads(cached)
return None
async def cache_related_entities(shout: Shout):
"""
Кэширует все связанные с публикацией сущности (авторов и темы)
"""
tasks = []
for author in shout.authors:
tasks.append(cache_by_id(Author, author.id, cache_author))
for topic in shout.topics:
tasks.append(cache_by_id(Topic, topic.id, cache_topic))
await asyncio.gather(*tasks)
async def invalidate_shout_related_cache(shout: Shout, author_id: int):
"""
Инвалидирует весь кэш, связанный с публикацией и её связями
Args:
shout: Объект публикации
author_id: ID автора
"""
cache_keys = {
"feed", # основная лента
f"author_{author_id}", # публикации автора
"random_top", # случайные топовые
"unrated", # неоцененные
"recent", # последние
"coauthored", # совместные
}
# Добавляем ключи авторов
cache_keys.update(f"author_{a.id}" for a in shout.authors)
cache_keys.update(f"authored_{a.id}" for a in shout.authors)
# Добавляем ключи тем
cache_keys.update(f"topic_{t.id}" for t in shout.topics)
cache_keys.update(f"topic_shouts_{t.id}" for t in shout.topics)
await invalidate_shouts_cache(list(cache_keys))
# Function removed - direct Redis calls used throughout the module instead
async def get_cached_entity(entity_type: str, entity_id: int, get_method, cache_method):
"""
Универсальная функция получения кэшированной сущности
Args:
entity_type: 'author' или 'topic'
entity_id: ID сущности
get_method: метод получения из БД
cache_method: метод кэширования
"""
key = f"{entity_type}:id:{entity_id}"
cached = await redis.execute("GET", key)
if cached:
return orjson.loads(cached)
entity = await get_method(entity_id)
if entity:
await cache_method(entity)
return entity
return None
async def cache_by_id(entity, entity_id: int, cache_method):
"""
Кэширует сущность по ID, используя указанный метод кэширования
Args:
entity: класс сущности (Author/Topic)
entity_id: ID сущности
cache_method: функция кэширования
"""
from resolvers.stat import get_with_stat
caching_query = select(entity).filter(entity.id == entity_id)
result = get_with_stat(caching_query)
if not result or not result[0]:
logger.warning(f"{entity.__name__} with id {entity_id} not found")
return
x = result[0]
d = x.dict()
await cache_method(d)
return d
# Универсальная функция для сохранения данных в кеш
async def cache_data(key: str, data: Any, ttl: Optional[int] = None) -> None:
"""
Сохраняет данные в кеш по указанному ключу.
Args:
key: Ключ кеша
data: Данные для сохранения
ttl: Время жизни кеша в секундах (None - бессрочно)
"""
try:
payload = json.dumps(data, cls=CustomJSONEncoder)
if ttl:
await redis.execute("SETEX", key, ttl, payload)
else:
await redis.execute("SET", key, payload)
logger.debug(f"Данные сохранены в кеш по ключу {key}")
except Exception as e:
logger.error(f"Ошибка при сохранении данных в кеш: {e}")
# Универсальная функция для получения данных из кеша
async def get_cached_data(key: str) -> Optional[Any]:
"""
Получает данные из кеша по указанному ключу.
Args:
key: Ключ кеша
Returns:
Any: Данные из кеша или None, если данных нет
"""
try:
cached_data = await redis.execute("GET", key)
if cached_data:
loaded = orjson.loads(cached_data)
logger.debug(f"Данные получены из кеша по ключу {key}: {len(loaded)}")
return loaded
return None
except Exception as e:
logger.error(f"Ошибка при получении данных из кеша: {e}")
return None
# Универсальная функция для инвалидации кеша по префиксу
async def invalidate_cache_by_prefix(prefix: str) -> None:
"""
Инвалидирует все ключи кеша с указанным префиксом.
Args:
prefix: Префикс ключей кеша для инвалидации
"""
try:
keys = await redis.execute("KEYS", f"{prefix}:*")
if keys:
await redis.execute("DEL", *keys)
logger.debug(f"Удалено {len(keys)} ключей кеша с префиксом {prefix}")
except Exception as e:
logger.error(f"Ошибка при инвалидации кеша: {e}")
# Универсальная функция для получения и кеширования данных
async def cached_query(
cache_key: str,
query_func: callable,
ttl: Optional[int] = None,
force_refresh: bool = False,
use_key_format: bool = True,
**query_params,
) -> Any:
"""
Gets data from cache or executes query and saves result to cache.
Supports existing key formats for compatibility.
Args:
cache_key: Cache key or key template from CACHE_KEYS
query_func: Function to execute the query
ttl: Cache TTL in seconds (None - indefinite)
force_refresh: Force cache refresh
use_key_format: Whether to check if cache_key matches a key template in CACHE_KEYS
**query_params: Parameters to pass to the query function
Returns:
Any: Data from cache or query result
"""
# Check if cache_key matches a pattern in CACHE_KEYS
actual_key = cache_key
if use_key_format and "{}" in cache_key:
# Look for a template match in CACHE_KEYS
for key_name, key_format in CACHE_KEYS.items():
if cache_key == key_format:
# We have a match, now look for the id or value to format with
for param_name, param_value in query_params.items():
if param_name in ["id", "slug", "user", "topic_id", "author_id"]:
actual_key = cache_key.format(param_value)
break
# If not forcing refresh, try to get data from cache
if not force_refresh:
cached_result = await get_cached_data(actual_key)
if cached_result is not None:
return cached_result
# If data not in cache or refresh required, execute query
try:
result = await query_func(**query_params)
if result is not None:
# Save result to cache
await cache_data(actual_key, result, ttl)
return result
except Exception as e:
logger.error(f"Error executing query for caching: {e}")
# In case of error, return data from cache if not forcing refresh
if not force_refresh:
return await get_cached_data(actual_key)
raise

11
cache/memorycache.py vendored Normal file
View File

@@ -0,0 +1,11 @@
from dogpile.cache import make_region
from settings import REDIS_URL
# Создание региона кэша с TTL
cache_region = make_region()
cache_region.configure(
"dogpile.cache.redis",
arguments={"url": f"{REDIS_URL}/1"},
expiration_time=3600, # Cache expiration time in seconds
)

14
cache/precache.py vendored
View File

@@ -86,15 +86,11 @@ async def precache_data():
# Преобразуем словарь в список аргументов для HSET
if value:
# Если значение - словарь, преобразуем его в плоский список для HSET
if isinstance(value, dict):
flattened = []
for field, val in value.items():
flattened.extend([field, val])
await redis.execute("HSET", key, *flattened)
else:
# Предполагаем, что значение уже содержит список
await redis.execute("HSET", key, *value)
flattened = []
for field, val in value.items():
flattened.extend([field, val])
await redis.execute("HSET", key, *flattened)
logger.info(f"redis hash '{key}' was restored")
with local_session() as session:

137
cache/revalidator.py vendored
View File

@@ -1,40 +1,20 @@
import asyncio
from cache.cache import (
cache_author,
cache_topic,
get_cached_author,
get_cached_topic,
invalidate_cache_by_prefix,
)
from cache.cache import cache_author, cache_topic, get_cached_author, get_cached_topic
from resolvers.stat import get_with_stat
from services.redis import redis
from utils.logger import root_logger as logger
CACHE_REVALIDATION_INTERVAL = 300 # 5 minutes
class CacheRevalidationManager:
def __init__(self, interval=CACHE_REVALIDATION_INTERVAL):
def __init__(self, interval=60):
"""Инициализация менеджера с заданным интервалом проверки (в секундах)."""
self.interval = interval
self.items_to_revalidate = {"authors": set(), "topics": set(), "shouts": set(), "reactions": set()}
self.lock = asyncio.Lock()
self.running = True
self.MAX_BATCH_SIZE = 10 # Максимальное количество элементов для поштучной обработки
self._redis = redis # Добавлена инициализация _redis для доступа к Redis-клиенту
async def start(self):
"""Запуск фонового воркера для ревалидации кэша."""
# Проверяем, что у нас есть соединение с Redis
if not self._redis._client:
logger.warning("Redis connection not established. Waiting for connection...")
try:
await self._redis.connect()
logger.info("Redis connection established for revalidation manager")
except Exception as e:
logger.error(f"Failed to connect to Redis: {e}")
self.task = asyncio.create_task(self.revalidate_cache())
async def revalidate_cache(self):
@@ -50,113 +30,24 @@ class CacheRevalidationManager:
async def process_revalidation(self):
"""Обновление кэша для всех сущностей, требующих ревалидации."""
# Проверяем соединение с Redis
if not self._redis._client:
return # Выходим из метода, если не удалось подключиться
async with self.lock:
# Ревалидация кэша авторов
if self.items_to_revalidate["authors"]:
logger.debug(f"Revalidating {len(self.items_to_revalidate['authors'])} authors")
for author_id in self.items_to_revalidate["authors"]:
if author_id == "all":
await invalidate_cache_by_prefix("authors")
break
author = await get_cached_author(author_id, get_with_stat)
if author:
await cache_author(author)
self.items_to_revalidate["authors"].clear()
for author_id in self.items_to_revalidate["authors"]:
author = await get_cached_author(author_id, get_with_stat)
if author:
await cache_author(author)
self.items_to_revalidate["authors"].clear()
# Ревалидация кэша тем
if self.items_to_revalidate["topics"]:
logger.debug(f"Revalidating {len(self.items_to_revalidate['topics'])} topics")
for topic_id in self.items_to_revalidate["topics"]:
if topic_id == "all":
await invalidate_cache_by_prefix("topics")
break
topic = await get_cached_topic(topic_id)
if topic:
await cache_topic(topic)
self.items_to_revalidate["topics"].clear()
# Ревалидация шаутов (публикаций)
if self.items_to_revalidate["shouts"]:
shouts_count = len(self.items_to_revalidate["shouts"])
logger.debug(f"Revalidating {shouts_count} shouts")
# Проверяем наличие специального флага 'all'
if "all" in self.items_to_revalidate["shouts"]:
await invalidate_cache_by_prefix("shouts")
# Если элементов много, но не 'all', используем специфический подход
elif shouts_count > self.MAX_BATCH_SIZE:
# Инвалидируем только collections keys, которые затрагивают много сущностей
collection_keys = await asyncio.create_task(self._redis.execute("KEYS", "shouts:*"))
if collection_keys:
await self._redis.execute("DEL", *collection_keys)
logger.debug(f"Удалено {len(collection_keys)} коллекционных ключей шаутов")
# Обновляем кеш каждого конкретного шаута
for shout_id in self.items_to_revalidate["shouts"]:
if shout_id != "all":
# Точечная инвалидация для каждого shout_id
specific_keys = [f"shout:id:{shout_id}"]
for key in specific_keys:
await self._redis.execute("DEL", key)
logger.debug(f"Удален ключ кеша {key}")
else:
# Если элементов немного, обрабатываем каждый
for shout_id in self.items_to_revalidate["shouts"]:
if shout_id != "all":
# Точечная инвалидация для каждого shout_id
specific_keys = [f"shout:id:{shout_id}"]
for key in specific_keys:
await self._redis.execute("DEL", key)
logger.debug(f"Удален ключ кеша {key}")
self.items_to_revalidate["shouts"].clear()
# Аналогично для реакций - точечная инвалидация
if self.items_to_revalidate["reactions"]:
reactions_count = len(self.items_to_revalidate["reactions"])
logger.debug(f"Revalidating {reactions_count} reactions")
if "all" in self.items_to_revalidate["reactions"]:
await invalidate_cache_by_prefix("reactions")
elif reactions_count > self.MAX_BATCH_SIZE:
# Инвалидируем только collections keys для реакций
collection_keys = await asyncio.create_task(self._redis.execute("KEYS", "reactions:*"))
if collection_keys:
await self._redis.execute("DEL", *collection_keys)
logger.debug(f"Удалено {len(collection_keys)} коллекционных ключей реакций")
# Точечная инвалидация для каждой реакции
for reaction_id in self.items_to_revalidate["reactions"]:
if reaction_id != "all":
specific_keys = [f"reaction:id:{reaction_id}"]
for key in specific_keys:
await self._redis.execute("DEL", key)
logger.debug(f"Удален ключ кеша {key}")
else:
# Точечная инвалидация для каждой реакции
for reaction_id in self.items_to_revalidate["reactions"]:
if reaction_id != "all":
specific_keys = [f"reaction:id:{reaction_id}"]
for key in specific_keys:
await self._redis.execute("DEL", key)
logger.debug(f"Удален ключ кеша {key}")
self.items_to_revalidate["reactions"].clear()
for topic_id in self.items_to_revalidate["topics"]:
topic = await get_cached_topic(topic_id)
if topic:
await cache_topic(topic)
self.items_to_revalidate["topics"].clear()
def mark_for_revalidation(self, entity_id, entity_type):
"""Отметить сущность для ревалидации."""
if entity_id and entity_type:
self.items_to_revalidate[entity_type].add(entity_id)
def invalidate_all(self, entity_type):
"""Пометить для инвалидации все элементы указанного типа."""
logger.debug(f"Marking all {entity_type} for invalidation")
# Особый флаг для полной инвалидации
self.items_to_revalidate[entity_type].add("all")
self.items_to_revalidate[entity_type].add(entity_id)
async def stop(self):
"""Остановка фонового воркера."""
@@ -169,4 +60,4 @@ class CacheRevalidationManager:
pass
revalidation_manager = CacheRevalidationManager()
revalidation_manager = CacheRevalidationManager(interval=300) # Ревалидация каждые 5 минут

88
cache/triggers.py vendored
View File

@@ -2,10 +2,9 @@ from sqlalchemy import event
from cache.revalidator import revalidation_manager
from orm.author import Author, AuthorFollower
from orm.reaction import Reaction, ReactionKind
from orm.reaction import Reaction
from orm.shout import Shout, ShoutAuthor, ShoutReactionsFollower
from orm.topic import Topic, TopicFollower
from services.db import local_session
from utils.logger import root_logger as logger
@@ -44,66 +43,6 @@ def after_follower_handler(mapper, connection, target, is_delete=False):
revalidation_manager.mark_for_revalidation(target.follower, "authors")
def after_shout_handler(mapper, connection, target):
"""Обработчик изменения статуса публикации"""
if not isinstance(target, Shout):
return
# Проверяем изменение статуса публикации
# was_published = target.published_at is not None and target.deleted_at is None
# Всегда обновляем счетчики для авторов и тем при любом изменении поста
for author in target.authors:
revalidation_manager.mark_for_revalidation(author.id, "authors")
for topic in target.topics:
revalidation_manager.mark_for_revalidation(topic.id, "topics")
# Обновляем сам пост
revalidation_manager.mark_for_revalidation(target.id, "shouts")
def after_reaction_handler(mapper, connection, target):
"""Обработчик для комментариев"""
if not isinstance(target, Reaction):
return
# Проверяем что это комментарий
is_comment = target.kind == ReactionKind.COMMENT.value
# Получаем связанный пост
shout_id = target.shout if isinstance(target.shout, int) else target.shout.id
if not shout_id:
return
# Обновляем счетчики для автора комментария
if target.created_by:
revalidation_manager.mark_for_revalidation(target.created_by, "authors")
# Обновляем счетчики для поста
revalidation_manager.mark_for_revalidation(shout_id, "shouts")
if is_comment:
# Для комментариев обновляем также авторов и темы
with local_session() as session:
shout = (
session.query(Shout)
.filter(
Shout.id == shout_id,
Shout.published_at.is_not(None),
Shout.deleted_at.is_(None),
)
.first()
)
if shout:
for author in shout.authors:
revalidation_manager.mark_for_revalidation(author.id, "authors")
for topic in shout.topics:
revalidation_manager.mark_for_revalidation(topic.id, "topics")
def events_register():
"""Регистрация обработчиков событий для всех сущностей."""
event.listen(ShoutAuthor, "after_insert", mark_for_revalidation)
@@ -112,36 +51,19 @@ def events_register():
event.listen(AuthorFollower, "after_insert", after_follower_handler)
event.listen(AuthorFollower, "after_update", after_follower_handler)
event.listen(
AuthorFollower,
"after_delete",
lambda *args: after_follower_handler(*args, is_delete=True),
)
event.listen(AuthorFollower, "after_delete", lambda *args: after_follower_handler(*args, is_delete=True))
event.listen(TopicFollower, "after_insert", after_follower_handler)
event.listen(TopicFollower, "after_update", after_follower_handler)
event.listen(
TopicFollower,
"after_delete",
lambda *args: after_follower_handler(*args, is_delete=True),
)
event.listen(TopicFollower, "after_delete", lambda *args: after_follower_handler(*args, is_delete=True))
event.listen(ShoutReactionsFollower, "after_insert", after_follower_handler)
event.listen(ShoutReactionsFollower, "after_update", after_follower_handler)
event.listen(
ShoutReactionsFollower,
"after_delete",
lambda *args: after_follower_handler(*args, is_delete=True),
)
event.listen(ShoutReactionsFollower, "after_delete", lambda *args: after_follower_handler(*args, is_delete=True))
event.listen(Reaction, "after_update", mark_for_revalidation)
event.listen(Author, "after_update", mark_for_revalidation)
event.listen(Topic, "after_update", mark_for_revalidation)
event.listen(Shout, "after_update", after_shout_handler)
event.listen(Shout, "after_delete", after_shout_handler)
event.listen(Reaction, "after_insert", after_reaction_handler)
event.listen(Reaction, "after_update", after_reaction_handler)
event.listen(Reaction, "after_delete", after_reaction_handler)
event.listen(Shout, "after_update", mark_for_revalidation)
logger.info("Event handlers registered successfully.")

295
docs/caching.md
View File

@@ -1,295 +0,0 @@
# Система кеширования Discours
## Общее описание
Система кеширования Discours - это комплексное решение для повышения производительности платформы. Она использует Redis для хранения часто запрашиваемых данных и уменьшения нагрузки на основную базу данных.
Кеширование реализовано как многоуровневая система, состоящая из нескольких модулей:
- `cache.py` - основной модуль с функциями кеширования
- `revalidator.py` - асинхронный менеджер ревалидации кеша
- `triggers.py` - триггеры событий SQLAlchemy для автоматической ревалидации
- `precache.py` - предварительное кеширование данных при старте приложения
## Ключевые компоненты
### 1. Форматы ключей кеша
Система поддерживает несколько форматов ключей для обеспечения совместимости и удобства использования:
- **Ключи сущностей**: `entity:property:value` (например, `author:id:123`)
- **Ключи коллекций**: `entity:collection:params` (например, `authors:stats:limit=10:offset=0`)
- **Специальные ключи**: для обратной совместимости (например, `topic_shouts_123`)
Все стандартные форматы ключей хранятся в словаре `CACHE_KEYS`:
```python
CACHE_KEYS = {
"TOPIC_ID": "topic:id:{}",
"TOPIC_SLUG": "topic:slug:{}",
"AUTHOR_ID": "author:id:{}",
# и другие...
}
```
### 2. Основные функции кеширования
#### Структура ключей
Вместо генерации ключей через вспомогательные функции, система следует строгим конвенциям формирования ключей:
1. **Ключи для отдельных сущностей** строятся по шаблону:
```
entity:property:value
```
Например:
- `topic:id:123` - тема с ID 123
- `author:slug:john-doe` - автор со слагом "john-doe"
- `shout:id:456` - публикация с ID 456
2. **Ключи для коллекций** строятся по шаблону:
```
entity:collection[:filter1=value1:filter2=value2:...]
```
Например:
- `topics:all:basic` - базовый список всех тем
- `authors:stats:limit=10:offset=0:sort=name` - отсортированный список авторов с пагинацией
- `shouts:feed:limit=20:community=1` - лента публикаций с фильтром по сообществу
3. **Специальные форматы ключей** для обратной совместимости:
```
entity_action_id
```
Например:
- `topic_shouts_123` - публикации для темы с ID 123
Во всех модулях системы разработчики должны явно формировать ключи в соответствии с этими конвенциями, что обеспечивает единообразие и предсказуемость кеширования.
#### Работа с данными в кеше
```python
async def cache_data(key, data, ttl=None)
async def get_cached_data(key)
```
Эти функции предоставляют универсальный интерфейс для сохранения и получения данных из кеша. Они напрямую используют Redis через вызовы `redis.execute()`.
#### Высокоуровневое кеширование запросов
```python
async def cached_query(cache_key, query_func, ttl=None, force_refresh=False, **query_params)
```
Функция `cached_query` объединяет получение данных из кеша и выполнение запроса в случае отсутствия данных в кеше. Это основная функция, которую следует использовать в резолверах для кеширования результатов запросов.
### 3. Кеширование сущностей
Для основных типов сущностей реализованы специальные функции:
```python
async def cache_topic(topic: dict)
async def cache_author(author: dict)
async def get_cached_topic(topic_id: int)
async def get_cached_author(author_id: int, get_with_stat)
```
Эти функции упрощают работу с часто используемыми типами данных и обеспечивают единообразный подход к их кешированию.
### 4. Работа со связями
Для работы со связями между сущностями предназначены функции:
```python
async def cache_follows(follower_id, entity_type, entity_id, is_insert=True)
async def get_cached_topic_followers(topic_id)
async def get_cached_author_followers(author_id)
async def get_cached_follower_topics(author_id)
```
Они позволяют эффективно кешировать и получать информацию о подписках, связях между авторами, темами и публикациями.
## Система инвалидации кеша
### 1. Прямая инвалидация
Система поддерживает два типа инвалидации кеша:
#### 1.1. Инвалидация по префиксу
```python
async def invalidate_cache_by_prefix(prefix)
```
Позволяет инвалидировать все ключи кеша, начинающиеся с указанного префикса. Используется в резолверах для инвалидации группы кешей при массовых изменениях.
#### 1.2. Точечная инвалидация
```python
async def invalidate_authors_cache(author_id=None)
async def invalidate_topics_cache(topic_id=None)
```
Эти функции позволяют инвалидировать кеш только для конкретной сущности, что снижает нагрузку на Redis и предотвращает ненужную потерю кешированных данных. Если ID сущности не указан, используется инвалидация по префиксу.
Примеры использования точечной инвалидации:
```python
# Инвалидация кеша только для автора с ID 123
await invalidate_authors_cache(123)
# Инвалидация кеша только для темы с ID 456
await invalidate_topics_cache(456)
```
### 2. Отложенная инвалидация
Модуль `revalidator.py` реализует систему отложенной инвалидации кеша через класс `CacheRevalidationManager`:
```python
class CacheRevalidationManager:
def __init__(self, interval=CACHE_REVALIDATION_INTERVAL):
# ...
self._redis = redis # Прямая ссылка на сервис Redis
async def start(self):
# Проверка и установка соединения с Redis
# ...
async def process_revalidation(self):
# Обработка элементов для ревалидации
# ...
def mark_for_revalidation(self, entity_id, entity_type):
# Добавляет сущность в очередь на ревалидацию
# ...
```
Менеджер ревалидации работает как асинхронный фоновый процесс, который периодически (по умолчанию каждые 5 минут) проверяет наличие сущностей для ревалидации.
**Взаимодействие с Redis:**
- CacheRevalidationManager хранит прямую ссылку на сервис Redis через атрибут `_redis`
- При запуске проверяется наличие соединения с Redis и при необходимости устанавливается новое
- Включена автоматическая проверка соединения перед каждой операцией ревалидации
- Система самостоятельно восстанавливает соединение при его потере
**Особенности реализации:**
- Для авторов и тем используется поштучная ревалидация каждой записи
- Для шаутов и реакций используется батчевая обработка, с порогом в 10 элементов
- При достижении порога система переключается на инвалидацию коллекций вместо поштучной обработки
- Специальный флаг `all` позволяет запустить полную инвалидацию всех записей типа
### 3. Автоматическая инвалидация через триггеры
Модуль `triggers.py` регистрирует обработчики событий SQLAlchemy, которые автоматически отмечают сущности для ревалидации при изменении данных в базе:
```python
def events_register():
event.listen(Author, "after_update", mark_for_revalidation)
event.listen(Topic, "after_update", mark_for_revalidation)
# и другие...
```
Триггеры имеют следующие особенности:
- Реагируют на события вставки, обновления и удаления
- Отмечают затронутые сущности для отложенной ревалидации
- Учитывают связи между сущностями (например, при изменении темы обновляются связанные шауты)
## Предварительное кеширование
Модуль `precache.py` реализует предварительное кеширование часто используемых данных при старте приложения:
```python
async def precache_data():
# ...
```
Эта функция выполняется при запуске приложения и заполняет кеш данными, которые будут часто запрашиваться пользователями.
## Примеры использования
### Простое кеширование результата запроса
```python
async def get_topics_with_stats(limit=10, offset=0, by="title"):
# Формирование ключа кеша по конвенции
cache_key = f"topics:stats:limit={limit}:offset={offset}:sort={by}"
cached_data = await get_cached_data(cache_key)
if cached_data:
return cached_data
# Выполнение запроса к базе данных
result = ... # логика получения данных
await cache_data(cache_key, result, ttl=300)
return result
```
### Использование обобщенной функции cached_query
```python
async def get_topics_with_stats(limit=10, offset=0, by="title"):
async def fetch_data(limit, offset, by):
# Логика получения данных
return result
# Формирование ключа кеша по конвенции
cache_key = f"topics:stats:limit={limit}:offset={offset}:sort={by}"
return await cached_query(
cache_key,
fetch_data,
ttl=300,
limit=limit,
offset=offset,
by=by
)
```
### Точечная инвалидация кеша при изменении данных
```python
async def update_topic(topic_id, new_data):
# Обновление данных в базе
# ...
# Точечная инвалидация кеша только для измененной темы
await invalidate_topics_cache(topic_id)
return updated_topic
```
## Отладка и мониторинг
Система кеширования использует логгер для отслеживания операций:
```python
logger.debug(f"Данные получены из кеша по ключу {key}")
logger.debug(f"Удалено {len(keys)} ключей кеша с префиксом {prefix}")
logger.error(f"Ошибка при инвалидации кеша: {e}")
```
Это позволяет отслеживать работу кеша и выявлять возможные проблемы на ранних стадиях.
## Рекомендации по использованию
1. **Следуйте конвенциям формирования ключей** - это критически важно для консистентности и предсказуемости кеша.
2. **Не создавайте собственные форматы ключей** - используйте существующие шаблоны для обеспечения единообразия.
3. **Не забывайте об инвалидации** - всегда инвалидируйте кеш при изменении данных.
4. **Используйте точечную инвалидацию** - вместо инвалидации по префиксу для снижения нагрузки на Redis.
5. **Устанавливайте разумные TTL** - используйте разные значения TTL в зависимости от частоты изменения данных.
6. **Не кешируйте большие объемы данных** - кешируйте только то, что действительно необходимо для повышения производительности.
## Технические детали реализации
- **Сериализация данных**: используется `orjson` для эффективной сериализации и десериализации данных.
- **Форматирование даты и времени**: для корректной работы с датами используется `CustomJSONEncoder`.
- **Асинхронность**: все операции кеширования выполняются асинхронно для минимального влияния на производительность API.
- **Прямое взаимодействие с Redis**: все операции выполняются через прямые вызовы `redis.execute()` с обработкой ошибок.
- **Батчевая обработка**: для массовых операций используется пороговое значение, после которого применяются оптимизированные стратегии.
## Известные ограничения
1. **Согласованность данных** - система не гарантирует абсолютную согласованность данных в кеше и базе данных.
2. **Память** - необходимо следить за объемом данных в кеше, чтобы избежать проблем с памятью Redis.
3. **Производительность Redis** - при большом количестве операций с кешем может стать узким местом.

165
docs/comments-pagination.md
View File

@@ -1,165 +0,0 @@
# Пагинация комментариев
## Обзор
Реализована система пагинации комментариев по веткам, которая позволяет эффективно загружать и отображать вложенные ветки обсуждений. Основные преимущества:
1. Загрузка только необходимых комментариев, а не всего дерева
2. Снижение нагрузки на сервер и клиент
3. Возможность эффективной навигации по большим обсуждениям
4. Предзагрузка первых N ответов для улучшения UX
## API для иерархической загрузки комментариев
### GraphQL запрос `load_comments_branch`
```graphql
query LoadCommentsBranch(
$shout: Int!,
$parentId: Int,
$limit: Int,
$offset: Int,
$sort: ReactionSort,
$childrenLimit: Int,
$childrenOffset: Int
) {
load_comments_branch(
shout: $shout,
parent_id: $parentId,
limit: $limit,
offset: $offset,
sort: $sort,
children_limit: $childrenLimit,
children_offset: $childrenOffset
) {
id
body
created_at
created_by {
id
name
slug
pic
}
kind
reply_to
stat {
rating
comments_count
}
first_replies {
id
body
created_at
created_by {
id
name
slug
pic
}
kind
reply_to
stat {
rating
comments_count
}
}
}
}
```
### Параметры запроса
| Параметр | Тип | По умолчанию | Описание |
|----------|-----|--------------|----------|
| shout | Int! | - | ID статьи, к которой относятся комментарии |
| parent_id | Int | null | ID родительского комментария. Если null, загружаются корневые комментарии |
| limit | Int | 10 | Максимальное количество комментариев для загрузки |
| offset | Int | 0 | Смещение для пагинации |
| sort | ReactionSort | newest | Порядок сортировки: newest, oldest, like |
| children_limit | Int | 3 | Максимальное количество дочерних комментариев для каждого родительского |
| children_offset | Int | 0 | Смещение для пагинации дочерних комментариев |
### Поля в ответе
Каждый комментарий содержит следующие основные поля:
- `id`: ID комментария
- `body`: Текст комментария
- `created_at`: Время создания
- `created_by`: Информация об авторе
- `kind`: Тип реакции (COMMENT)
- `reply_to`: ID родительского комментария (null для корневых)
- `first_replies`: Первые N дочерних комментариев
- `stat`: Статистика комментария, включающая:
- `comments_count`: Количество ответов на комментарий
- `rating`: Рейтинг комментария
## Примеры использования
### Загрузка корневых комментариев с первыми ответами
```javascript
const { data } = await client.query({
query: LOAD_COMMENTS_BRANCH,
variables: {
shout: 222,
limit: 10,
offset: 0,
sort: "newest",
childrenLimit: 3
}
});
```
### Загрузка ответов на конкретный комментарий
```javascript
const { data } = await client.query({
query: LOAD_COMMENTS_BRANCH,
variables: {
shout: 222,
parentId: 123, // ID комментария, для которого загружаем ответы
limit: 10,
offset: 0,
sort: "oldest" // Сортируем ответы от старых к новым
}
});
```
### Пагинация дочерних комментариев
Для загрузки дополнительных ответов на комментарий:
```javascript
const { data } = await client.query({
query: LOAD_COMMENTS_BRANCH,
variables: {
shout: 222,
parentId: 123,
limit: 10,
offset: 0,
childrenLimit: 5,
childrenOffset: 3 // Пропускаем первые 3 комментария (уже загруженные)
}
});
```
## Рекомендации по клиентской реализации
1. Для эффективной работы со сложными ветками обсуждений рекомендуется:
- Сначала загружать только корневые комментарии с первыми N ответами
- При наличии дополнительных ответов (когда `stat.comments_count > first_replies.length`)
добавить кнопку "Показать все ответы"
- При нажатии на кнопку загружать дополнительные ответы с помощью запроса с указанным `parentId`
2. Для сортировки:
- По умолчанию использовать `newest` для отображения свежих обсуждений
- Предусмотреть переключатель сортировки для всего дерева комментариев
- При изменении сортировки перезагружать данные с новым параметром `sort`
3. Для улучшения производительности:
- Кешировать результаты запросов на клиенте
- Использовать оптимистичные обновления при добавлении/редактировании комментариев
- При необходимости загружать комментарии порциями (ленивая загрузка)

33
docs/features.md
View File

@@ -6,20 +6,14 @@
## Мультидоменная авторизация
- Поддержка авторизации для разных доменов
- Поддержка авторизации для разных доменов:
- *.dscrs.site (включая testing.dscrs.site)
- localhost[:port]
- testingdiscoursio-git-*-discoursio.vercel.app
- *.discours.io
- Автоматическое определение сервера авторизации
- Корректная обработка CORS для всех поддерживаемых доменов
## Система кеширования
- Redis используется в качестве основного механизма кеширования
- Поддержка как синхронных, так и асинхронных функций в декораторе cache_on_arguments
- Автоматическая сериализация/десериализация данных в JSON с использованием CustomJSONEncoder
- Резервная сериализация через pickle для сложных объектов
- Генерация уникальных ключей кеша на основе сигнатуры функции и переданных аргументов
- Настраиваемое время жизни кеша (TTL)
- Возможность ручной инвалидации кеша для конкретных функций и аргументов
## Webhooks
- Автоматическая регистрация вебхука для события user.login
@@ -31,18 +25,11 @@
## CORS Configuration
- Поддерживаются домены:
- *.dscrs.site (включая testing.dscrs.site, core.dscrs.site)
- *.discours.io (включая testing.discours.io)
- localhost (включая порты)
- Поддерживаемые методы: GET, POST, OPTIONS
- Настроена поддержка credentials
- Разрешенные заголовки: Authorization, Content-Type, X-Requested-With, DNT, Cache-Control
- Настроено кэширование preflight-ответов на 20 дней (1728000 секунд)
## Пагинация комментариев по веткам
- Эффективная загрузка комментариев с учетом их иерархической структуры
- Отдельный запрос `load_comments_branch` для оптимизированной загрузки ветки комментариев
- Возможность загрузки корневых комментариев статьи с первыми ответами на них
- Гибкая пагинация как для корневых, так и для дочерних комментариев
- Использование поля `stat.comments_count` для отображения количества ответов на комментарий
- Добавление специального поля `first_replies` для хранения первых ответов на комментарий
- Поддержка различных методов сортировки (новые, старые, популярные)
- Оптимизированные SQL запросы для минимизации нагрузки на базу данных
- Настроено кэширование preflight-ответов на 20 дней (1728000 секунд)

46
main.py
View File

@@ -14,9 +14,20 @@ from starlette.routing import Route
from cache.precache import precache_data
from cache.revalidator import revalidation_manager
from orm import (
# collection,
# invite,
author,
community,
notification,
reaction,
shout,
topic,
)
from services.db import create_table_if_not_exists, engine
from services.exception import ExceptionHandlerMiddleware
from services.redis import redis
from services.schema import create_all_tables, resolvers
from services.schema import resolvers
from services.search import search_service
from services.viewed import ViewedStorage
from services.webhook import WebhookEndpoint, create_webhook_endpoint
@@ -35,10 +46,39 @@ async def start():
print(f"[main] process started in {MODE} mode")
async def lifespan(_app):
def create_all_tables():
for model in [
# user.User,
author.Author,
author.AuthorFollower,
community.Community,
community.CommunityFollower,
shout.Shout,
shout.ShoutAuthor,
author.AuthorBookmark,
topic.Topic,
topic.TopicFollower,
shout.ShoutTopic,
reaction.Reaction,
shout.ShoutReactionsFollower,
author.AuthorRating,
notification.Notification,
notification.NotificationSeen,
# collection.Collection, collection.ShoutCollection,
# invite.Invite
]:
create_table_if_not_exists(engine, model)
async def create_all_tables_async():
# Оборачиваем синхронную функцию в асинхронную
await asyncio.to_thread(create_all_tables)
async def lifespan(app):
try:
create_all_tables()
await asyncio.gather(
create_all_tables_async(),
redis.connect(),
precache_data(),
ViewedStorage.init(),

71
nginx.conf.sigil
View File

@@ -1,11 +1,16 @@
log_format custom '$remote_addr - $remote_user [$time_local] "$request" '
'origin=$http_origin allow_origin=$allow_origin status=$status '
'"$http_referer" "$http_user_agent"';
{{ $proxy_settings := "proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection $http_connection; proxy_set_header Host $http_host; proxy_set_header X-Request-Start $msec;" }}
{{ $gzip_settings := "gzip on; gzip_min_length 1100; gzip_buffers 4 32k; gzip_types text/css text/javascript text/xml text/plain text/x-component application/javascript application/x-javascript application/json application/xml application/rss+xml font/truetype application/x-font-ttf font/opentype application/vnd.ms-fontobject image/svg+xml; gzip_vary on; gzip_comp_level 6;" }}
proxy_cache_path /var/cache/nginx levels=1:2 keys_zone=my_cache:10m max_size=1g
{{ $cors_headers_options := "if ($request_method = 'OPTIONS') { add_header 'Access-Control-Allow-Origin' '$allow_origin' always; add_header 'Access-Control-Allow-Methods' 'POST, GET, OPTIONS'; add_header 'Access-Control-Allow-Headers' 'Content-Type, Authorization' always; add_header 'Access-Control-Allow-Credentials' 'true' always; add_header 'Access-Control-Max-Age' 1728000; add_header 'Content-Type' 'text/plain; charset=utf-8'; add_header 'Content-Length' 0; return 204; }" }}
{{ $cors_headers_post := "if ($request_method = 'POST') { add_header 'Access-Control-Allow-Origin' '$allow_origin' always; add_header 'Access-Control-Allow-Methods' 'POST, GET, OPTIONS' always; add_header 'Access-Control-Allow-Headers' 'Content-Type, Authorization' always; add_header 'Access-Control-Allow-Credentials' 'true' always; }" }}
{{ $cors_headers_get := "if ($request_method = 'GET') { add_header 'Access-Control-Allow-Origin' '$allow_origin' always; add_header 'Access-Control-Allow-Methods' 'POST, GET, OPTIONS' always; add_header 'Access-Control-Allow-Headers' 'Content-Type, Authorization' always; add_header 'Access-Control-Allow-Credentials' 'true' always; }" }}
map $http_origin $allow_origin {
~^https?:\/\/((.*\.)?localhost(:\d+)?|discoursio-webapp(-(.*))?\.vercel\.app|(.*\.)?discours\.io|(.*\.)?dscrs\.site)$ $http_origin;
default "";
}
proxy_cache_path /var/cache/nginx levels=1:2 keys_zone=my_cache:10m max_size=1g
inactive=60m use_temp_path=off;
limit_conn_zone $binary_remote_addr zone=addr:10m;
limit_req_zone $binary_remote_addr zone=req_zone:10m rate=20r/s;
@@ -21,7 +26,7 @@ server {
listen [::]:{{ $listen_port }};
listen {{ $listen_port }};
server_name {{ $.NOSSL_SERVER_NAME }};
access_log /var/log/nginx/{{ $.APP }}-access.log custom;
access_log /var/log/nginx/{{ $.APP }}-access.log;
error_log /var/log/nginx/{{ $.APP }}-error.log;
client_max_body_size 100M;
@@ -29,7 +34,7 @@ server {
listen [::]:{{ $listen_port }} ssl http2;
listen {{ $listen_port }} ssl http2;
server_name {{ $.NOSSL_SERVER_NAME }};
access_log /var/log/nginx/{{ $.APP }}-access.log custom;
access_log /var/log/nginx/{{ $.APP }}-access.log;
error_log /var/log/nginx/{{ $.APP }}-error.log;
ssl_certificate {{ $.APP_SSL_PATH }}/server.crt;
ssl_certificate_key {{ $.APP_SSL_PATH }}/server.key;
@@ -45,37 +50,12 @@ server {
location / {
proxy_pass http://{{ $.APP }}-{{ $upstream_port }};
proxy_pass http://{{ $.APP }}-{{ $upstream_port }};
{{ $proxy_settings }}
{{ $gzip_settings }}
# Handle CORS for OPTIONS method
if ($request_method = 'OPTIONS') {
add_header 'Access-Control-Allow-Origin' $allow_origin always;
add_header 'Access-Control-Allow-Methods' 'POST, GET, OPTIONS';
add_header 'Access-Control-Allow-Headers' 'Content-Type, Authorization' always;
add_header 'Access-Control-Allow-Credentials' 'true' always;
add_header 'Access-Control-Max-Age' 1728000;
add_header 'Content-Type' 'text/plain; charset=utf-8';
add_header 'Content-Length' 0;
return 204;
}
# Handle CORS for POST method
if ($request_method = 'POST') {
add_header 'Access-Control-Allow-Origin' $allow_origin always;
add_header 'Access-Control-Allow-Methods' 'POST, GET, OPTIONS' always;
add_header 'Access-Control-Allow-Headers' 'Content-Type, Authorization' always;
add_header 'Access-Control-Allow-Credentials' 'true' always;
}
# Handle CORS for GET method
if ($request_method = 'GET') {
add_header 'Access-Control-Allow-Origin' $allow_origin always;
add_header 'Access-Control-Allow-Methods' 'POST, GET, OPTIONS' always;
add_header 'Access-Control-Allow-Headers' 'Content-Type, Authorization' always;
add_header 'Access-Control-Allow-Credentials' 'true' always;
}
{{ $cors_headers_options }}
{{ $cors_headers_post }}
{{ $cors_headers_get }}
proxy_cache my_cache;
proxy_cache_revalidate on;
@@ -85,17 +65,26 @@ server {
proxy_cache_lock on;
# Connections and request limits increase (bad for DDos)
limit_conn addr 10000;
limit_req zone=req_zone burst=10 nodelay;
}
# Custom location block for /upload
# location /upload {
# proxy_pass http://uploader-8080/;
# {{ $proxy_settings }}
# {{ $gzip_settings }}
# {{ $cors_headers_options }}
# {{ $cors_headers_post }}
# {{ $cors_headers_get }}
# }
location ~* \.(jpg|jpeg|png|gif|ico|css|js)$ {
proxy_pass http://{{ $.APP }}-{{ $upstream_port }};
expires 30d;
add_header Cache-Control "public, no-transform";
expires 30d; # This means that the client can cache these resources for 30 days.
add_header Cache-Control "public, no-transform";
}
location ~* \.(mp3|wav|ogg|flac|aac|aif|webm)$ {
proxy_pass http://{{ $.APP }}-{{ $upstream_port }};
location ~* \.(mp3)$ {
if ($request_method = 'GET') {
add_header 'Access-Control-Allow-Origin' $allow_origin always;
add_header 'Access-Control-Allow-Methods' 'GET, POST, OPTIONS' always;

87
orm/author.py
View File

@@ -1,6 +1,6 @@
import time
from sqlalchemy import JSON, Boolean, Column, ForeignKey, Index, Integer, String
from sqlalchemy import JSON, Boolean, Column, ForeignKey, Integer, String
from services.db import Base
@@ -8,15 +8,6 @@ from services.db import Base
class AuthorRating(Base):
"""
Рейтинг автора от другого автора.
Attributes:
rater (int): ID оценивающего автора
author (int): ID оцениваемого автора
plus (bool): Положительная/отрицательная оценка
"""
__tablename__ = "author_rating"
id = None # type: ignore
@@ -24,26 +15,8 @@ class AuthorRating(Base):
author = Column(ForeignKey("author.id"), primary_key=True)
plus = Column(Boolean)
# Определяем индексы
__table_args__ = (
# Индекс для быстрого поиска всех оценок конкретного автора
Index("idx_author_rating_author", "author"),
# Индекс для быстрого поиска всех оценок, оставленных конкретным автором
Index("idx_author_rating_rater", "rater"),
)
class AuthorFollower(Base):
"""
Подписка одного автора на другого.
Attributes:
follower (int): ID подписчика
author (int): ID автора, на которого подписываются
created_at (int): Время создания подписки
auto (bool): Признак автоматической подписки
"""
__tablename__ = "author_follower"
id = None # type: ignore
@@ -52,58 +25,20 @@ class AuthorFollower(Base):
created_at = Column(Integer, nullable=False, default=lambda: int(time.time()))
auto = Column(Boolean, nullable=False, default=False)
# Определяем индексы
__table_args__ = (
# Индекс для быстрого поиска всех подписчиков автора
Index("idx_author_follower_author", "author"),
# Индекс для быстрого поиска всех авторов, на которых подписан конкретный автор
Index("idx_author_follower_follower", "follower"),
)
class AuthorBookmark(Base):
"""
Закладка автора на публикацию.
Attributes:
author (int): ID автора
shout (int): ID публикации
"""
__tablename__ = "author_bookmark"
id = None # type: ignore
author = Column(ForeignKey("author.id"), primary_key=True)
shout = Column(ForeignKey("shout.id"), primary_key=True)
# Определяем индексы
__table_args__ = (
# Индекс для быстрого поиска всех закладок автора
Index("idx_author_bookmark_author", "author"),
# Индекс для быстрого поиска всех авторов, добавивших публикацию в закладки
Index("idx_author_bookmark_shout", "shout"),
)
class Author(Base):
"""
Модель автора в системе.
Attributes:
name (str): Отображаемое имя
slug (str): Уникальный строковый идентификатор
bio (str): Краткая биография/статус
about (str): Полное описание
pic (str): URL изображения профиля
links (dict): Ссылки на социальные сети и сайты
created_at (int): Время создания профиля
last_seen (int): Время последнего посещения
updated_at (int): Время последнего обновления
deleted_at (int): Время удаления (если профиль удален)
"""
__tablename__ = "author"
user = Column(String) # unbounded link with authorizer's User type
name = Column(String, nullable=True, comment="Display name")
slug = Column(String, unique=True, comment="Author's slug")
bio = Column(String, nullable=True, comment="Bio") # status description
@@ -118,19 +53,3 @@ class Author(Base):
# search_vector = Column(
# TSVectorType("name", "slug", "bio", "about", regconfig="pg_catalog.russian")
# )
# Определяем индексы
__table_args__ = (
# Индекс для быстрого поиска по имени
Index("idx_author_name", "name"),
# Индекс для быстрого поиска по slug
Index("idx_author_slug", "slug"),
# Индекс для фильтрации неудаленных авторов
Index(
"idx_author_deleted_at", "deleted_at", postgresql_where=deleted_at.is_(None)
),
# Индекс для сортировки по времени создания (для новых авторов)
Index("idx_author_created_at", "created_at"),
# Индекс для сортировки по времени последнего посещения
Index("idx_author_last_seen", "last_seen"),
)

105
orm/draft.py
View File

@@ -1,105 +0,0 @@
import time
from sqlalchemy import JSON, Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from orm.author import Author
from orm.topic import Topic
from services.db import Base
class DraftTopic(Base):
__tablename__ = "draft_topic"
id = None # type: ignore
shout = Column(ForeignKey("draft.id"), primary_key=True, index=True)
topic = Column(ForeignKey("topic.id"), primary_key=True, index=True)
main = Column(Boolean, nullable=True)
class DraftAuthor(Base):
__tablename__ = "draft_author"
id = None # type: ignore
shout = Column(ForeignKey("draft.id"), primary_key=True, index=True)
author = Column(ForeignKey("author.id"), primary_key=True, index=True)
caption = Column(String, nullable=True, default="")
class Draft(Base):
__tablename__ = "draft"
# required
created_at: int = Column(Integer, nullable=False, default=lambda: int(time.time()))
# Колонки для связей с автором
created_by: int = Column("created_by", ForeignKey("author.id"), nullable=False)
community: int = Column("community", ForeignKey("community.id"), nullable=False, default=1)
# optional
layout: str = Column(String, nullable=True, default="article")
slug: str = Column(String, unique=True)
title: str = Column(String, nullable=True)
subtitle: str | None = Column(String, nullable=True)
lead: str | None = Column(String, nullable=True)
body: str = Column(String, nullable=False, comment="Body")
media: dict | None = Column(JSON, nullable=True)
cover: str | None = Column(String, nullable=True, comment="Cover image url")
cover_caption: str | None = Column(String, nullable=True, comment="Cover image alt caption")
lang: str = Column(String, nullable=False, default="ru", comment="Language")
seo: str | None = Column(String, nullable=True) # JSON
# auto
updated_at: int | None = Column(Integer, nullable=True, index=True)
deleted_at: int | None = Column(Integer, nullable=True, index=True)
updated_by: int | None = Column("updated_by", ForeignKey("author.id"), nullable=True)
deleted_by: int | None = Column("deleted_by", ForeignKey("author.id"), nullable=True)
# --- Relationships ---
# Только many-to-many связи через вспомогательные таблицы
authors = relationship(Author, secondary="draft_author", lazy="select")
topics = relationship(Topic, secondary="draft_topic", lazy="select")
# Связь с Community (если нужна как объект, а не ID)
# community = relationship("Community", foreign_keys=[community_id], lazy="joined")
# Пока оставляем community_id как ID
# Связь с публикацией (один-к-одному или один-к-нулю)
# Загружается через joinedload в резолвере
publication = relationship(
"Shout",
primaryjoin="Draft.id == Shout.draft",
foreign_keys="Shout.draft",
uselist=False,
lazy="noload", # Не грузим по умолчанию, только через options
viewonly=True # Указываем, что это связь только для чтения
)
def dict(self):
"""
Сериализует объект Draft в словарь.
Гарантирует, что поля topics и authors всегда будут списками.
"""
return {
"id": self.id,
"created_at": self.created_at,
"created_by": self.created_by,
"community": self.community,
"layout": self.layout,
"slug": self.slug,
"title": self.title,
"subtitle": self.subtitle,
"lead": self.lead,
"body": self.body,
"media": self.media or [],
"cover": self.cover,
"cover_caption": self.cover_caption,
"lang": self.lang,
"seo": self.seo,
"updated_at": self.updated_at,
"deleted_at": self.deleted_at,
"updated_by": self.updated_by,
"deleted_by": self.deleted_by,
# Гарантируем, что topics и authors всегда будут списками
"topics": [topic.dict() for topic in (self.topics or [])],
"authors": [author.dict() for author in (self.authors or [])]
}

100
orm/shout.py
View File

@@ -1,6 +1,6 @@
import time
from sqlalchemy import JSON, Boolean, Column, ForeignKey, Index, Integer, String
from sqlalchemy import JSON, Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from orm.author import Author
@@ -10,15 +10,6 @@ from services.db import Base
class ShoutTopic(Base):
"""
Связь между публикацией и темой.
Attributes:
shout (int): ID публикации
topic (int): ID темы
main (bool): Признак основной темы
"""
__tablename__ = "shout_topic"
id = None # type: ignore
@@ -26,12 +17,6 @@ class ShoutTopic(Base):
topic = Column(ForeignKey("topic.id"), primary_key=True, index=True)
main = Column(Boolean, nullable=True)
# Определяем дополнительные индексы
__table_args__ = (
# Оптимизированный составной индекс для запросов, которые ищут публикации по теме
Index("idx_shout_topic_topic_shout", "topic", "shout"),
)
class ShoutReactionsFollower(Base):
__tablename__ = "shout_reactions_followers"
@@ -45,15 +30,6 @@ class ShoutReactionsFollower(Base):
class ShoutAuthor(Base):
"""
Связь между публикацией и автором.
Attributes:
shout (int): ID публикации
author (int): ID автора
caption (str): Подпись автора
"""
__tablename__ = "shout_author"
id = None # type: ignore
@@ -61,66 +37,38 @@ class ShoutAuthor(Base):
author = Column(ForeignKey("author.id"), primary_key=True, index=True)
caption = Column(String, nullable=True, default="")
# Определяем дополнительные индексы
__table_args__ = (
# Оптимизированный индекс для запросов, которые ищут публикации по автору
Index("idx_shout_author_author_shout", "author", "shout"),
)
class Shout(Base):
"""
Публикация в системе.
"""
__tablename__ = "shout"
created_at: int = Column(Integer, nullable=False, default=lambda: int(time.time()))
updated_at: int | None = Column(Integer, nullable=True, index=True)
published_at: int | None = Column(Integer, nullable=True, index=True)
featured_at: int | None = Column(Integer, nullable=True, index=True)
deleted_at: int | None = Column(Integer, nullable=True, index=True)
created_at = Column(Integer, nullable=False, default=lambda: int(time.time()))
updated_at = Column(Integer, nullable=True, index=True)
published_at = Column(Integer, nullable=True, index=True)
featured_at = Column(Integer, nullable=True, index=True)
deleted_at = Column(Integer, nullable=True, index=True)
created_by: int = Column(ForeignKey("author.id"), nullable=False)
updated_by: int | None = Column(ForeignKey("author.id"), nullable=True)
deleted_by: int | None = Column(ForeignKey("author.id"), nullable=True)
community: int = Column(ForeignKey("community.id"), nullable=False)
created_by = Column(ForeignKey("author.id"), nullable=False)
updated_by = Column(ForeignKey("author.id"), nullable=True)
deleted_by = Column(ForeignKey("author.id"), nullable=True)
community = Column(ForeignKey("community.id"), nullable=False)
body: str = Column(String, nullable=False, comment="Body")
slug: str = Column(String, unique=True)
cover: str | None = Column(String, nullable=True, comment="Cover image url")
cover_caption: str | None = Column(String, nullable=True, comment="Cover image alt caption")
lead: str | None = Column(String, nullable=True)
title: str = Column(String, nullable=False)
subtitle: str | None = Column(String, nullable=True)
layout: str = Column(String, nullable=False, default="article")
media: dict | None = Column(JSON, nullable=True)
body = Column(String, nullable=False, comment="Body")
slug = Column(String, unique=True)
cover = Column(String, nullable=True, comment="Cover image url")
cover_caption = Column(String, nullable=True, comment="Cover image alt caption")
lead = Column(String, nullable=True)
description = Column(String, nullable=True)
title = Column(String, nullable=False)
subtitle = Column(String, nullable=True)
layout = Column(String, nullable=False, default="article")
media = Column(JSON, nullable=True)
authors = relationship(Author, secondary="shout_author")
topics = relationship(Topic, secondary="shout_topic")
reactions = relationship(Reaction)
lang: str = Column(String, nullable=False, default="ru", comment="Language")
version_of: int | None = Column(ForeignKey("shout.id"), nullable=True)
oid: str | None = Column(String, nullable=True)
lang = Column(String, nullable=False, default="ru", comment="Language")
version_of = Column(ForeignKey("shout.id"), nullable=True)
oid = Column(String, nullable=True)
seo: str | None = Column(String, nullable=True) # JSON
draft: int | None = Column(ForeignKey("draft.id"), nullable=True)
# Определяем индексы
__table_args__ = (
# Индекс для быстрого поиска неудаленных публикаций
Index("idx_shout_deleted_at", "deleted_at", postgresql_where=deleted_at.is_(None)),
# Индекс для быстрой фильтрации по community
Index("idx_shout_community", "community"),
# Индекс для быстрого поиска по slug
Index("idx_shout_slug", "slug"),
# Составной индекс для фильтрации опубликованных неудаленных публикаций
Index(
"idx_shout_published_deleted",
"published_at",
"deleted_at",
postgresql_where=published_at.is_not(None) & deleted_at.is_(None),
),
)
seo = Column(String, nullable=True) # JSON

42
orm/topic.py
View File

@@ -1,21 +1,11 @@
import time
from sqlalchemy import JSON, Boolean, Column, ForeignKey, Index, Integer, String
from sqlalchemy import JSON, Boolean, Column, ForeignKey, Integer, String
from services.db import Base
class TopicFollower(Base):
"""
Связь между топиком и его подписчиком.
Attributes:
follower (int): ID подписчика
topic (int): ID топика
created_at (int): Время создания связи
auto (bool): Автоматическая подписка
"""
__tablename__ = "topic_followers"
id = None # type: ignore
@@ -24,29 +14,8 @@ class TopicFollower(Base):
created_at = Column(Integer, nullable=False, default=int(time.time()))
auto = Column(Boolean, nullable=False, default=False)
# Определяем индексы
__table_args__ = (
# Индекс для быстрого поиска всех подписчиков топика
Index("idx_topic_followers_topic", "topic"),
# Индекс для быстрого поиска всех топиков, на которые подписан автор
Index("idx_topic_followers_follower", "follower"),
)
class Topic(Base):
"""
Модель топика (темы) публикаций.
Attributes:
slug (str): Уникальный строковый идентификатор темы
title (str): Название темы
body (str): Описание темы
pic (str): URL изображения темы
community (int): ID сообщества
oid (str): Старый ID
parent_ids (list): IDs родительских тем
"""
__tablename__ = "topic"
slug = Column(String, unique=True)
@@ -55,12 +24,5 @@ class Topic(Base):
pic = Column(String, nullable=True, comment="Picture")
community = Column(ForeignKey("community.id"), default=1)
oid = Column(String, nullable=True, comment="Old ID")
parent_ids = Column(JSON, nullable=True, comment="Parent Topic IDs")
# Определяем индексы
__table_args__ = (
# Индекс для быстрого поиска по slug
Index("idx_topic_slug", "slug"),
# Индекс для быстрого поиска по сообществу
Index("idx_topic_community", "community"),
)
parent_ids = Column(JSON, nullable=True, comment="Parent Topic IDs")

51
pyproject.toml
View File

@@ -1,2 +1,51 @@
[tool.poetry]
name = "core"
version = "0.4.7"
description = "core module for discours.io"
authors = ["discoursio devteam"]
license = "MIT"
readme = "README.md"
[tool.poetry.dependencies]
python = "^3.12"
SQLAlchemy = "^2.0.29"
psycopg2-binary = "^2.9.9"
redis = {extras = ["hiredis"], version = "^5.0.1"}
sentry-sdk = {version = "^1.44.1", extras = ["starlette", "ariadne", "sqlalchemy"]}
starlette = "^0.39.2"
gql = "^3.5.0"
ariadne = "^0.23.0"
pre-commit = "^3.7.0"
granian = "^1.4.1"
google-analytics-data = "^0.18.7"
opensearch-py = "^2.6.0"
httpx = "^0.27.0"
dogpile-cache = "^1.3.1"
colorlog = "^6.8.2"
fakeredis = "^2.25.1"
pydantic = "^2.9.2"
jwt = "^1.3.1"
authlib = "^1.3.2"
[tool.poetry.group.dev.dependencies]
ruff = "^0.4.7"
isort = "^5.13.2"
pydantic = "^2.9.2"
[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"
[tool.pyright]
venvPath = "."
venv = ".venv"
[tool.isort]
multi_line_output = 3
include_trailing_comma = true
force_grid_wrap = 0
line_length = 120
[tool.ruff]
line-length = 108
line-length = 120

6
requirements.dev.txt
View File

@@ -1,6 +0,0 @@
fakeredis
pytest
pytest-asyncio
pytest-cov
mypy
ruff

18
requirements.txt
View File

@@ -1,18 +0,0 @@
# own auth
bcrypt
authlib
passlib
opensearch-py
google-analytics-data
colorlog
psycopg2-binary
httpx
redis[hiredis]
sentry-sdk[starlette,sqlalchemy]
starlette
gql
ariadne
granian
orjson
pydantic
trafilatura

26
resolvers/__init__.py
View File

@@ -11,16 +11,7 @@ from resolvers.author import ( # search_authors,
update_author,
)
from resolvers.community import get_communities_all, get_community
from resolvers.draft import (
create_draft,
delete_draft,
load_drafts,
publish_draft,
update_draft,
)
from resolvers.editor import (
unpublish_shout,
)
from resolvers.editor import create_shout, delete_shout, update_shout
from resolvers.feed import (
load_shouts_coauthored,
load_shouts_discussed,
@@ -39,7 +30,6 @@ from resolvers.reaction import (
create_reaction,
delete_reaction,
load_comment_ratings,
load_comments_branch,
load_reactions_by,
load_shout_comments,
load_shout_ratings,
@@ -102,6 +92,10 @@ __all__ = [
"follow",
"unfollow",
"get_shout_followers",
# editor
"create_shout",
"update_shout",
"delete_shout",
# reaction
"create_reaction",
"update_reaction",
@@ -110,7 +104,6 @@ __all__ = [
"load_shout_comments",
"load_shout_ratings",
"load_comment_ratings",
"load_comments_branch",
# notifier
"load_notifications",
"notifications_seen_thread",
@@ -120,13 +113,4 @@ __all__ = [
"rate_author",
"get_my_rates_comments",
"get_my_rates_shouts",
# draft
"load_drafts",
"create_draft",
"update_draft",
"delete_draft",
"publish_draft",
"publish_shout",
"unpublish_shout",
"unpublish_draft",
]

441
resolvers/author.py
View File

@@ -1,272 +1,25 @@
import asyncio
import time
from typing import Optional
from sqlalchemy import select, text
from sqlalchemy import desc, select, text
from cache.cache import (
cache_author,
cached_query,
get_cached_author,
get_cached_author_by_user_id,
get_cached_author_followers,
get_cached_follower_authors,
get_cached_follower_topics,
invalidate_cache_by_prefix,
)
from orm.author import Author
from orm.shout import ShoutAuthor, ShoutTopic
from orm.topic import Topic
from resolvers.stat import get_with_stat
from services.auth import login_required
from services.db import local_session
from services.redis import redis
from services.schema import mutation, query
from utils.logger import root_logger as logger
DEFAULT_COMMUNITIES = [1]
# Вспомогательная функция для получения всех авторов без статистики
async def get_all_authors():
"""
Получает всех авторов без статистики.
Используется для случаев, когда нужен полный список авторов без дополнительной информации.
Returns:
list: Список всех авторов без статистики
"""
cache_key = "authors:all:basic"
# Функция для получения всех авторов из БД
async def fetch_all_authors():
logger.debug("Получаем список всех авторов из БД и кешируем результат")
with local_session() as session:
# Запрос на получение базовой информации об авторах
authors_query = select(Author).where(Author.deleted_at.is_(None))
authors = session.execute(authors_query).scalars().all()
# Преобразуем авторов в словари
return [author.dict() for author in authors]
# Используем универсальную функцию для кеширования запросов
return await cached_query(cache_key, fetch_all_authors)
# Вспомогательная функция для получения авторов со статистикой с пагинацией
async def get_authors_with_stats(limit=50, offset=0, by: Optional[str] = None):
"""
Получает авторов со статистикой с пагинацией.
Args:
limit: Максимальное количество возвращаемых авторов
offset: Смещение для пагинации
by: Опциональный параметр сортировки (new/active)
Returns:
list: Список авторов с их статистикой
"""
# Формируем ключ кеша с помощью универсальной функции
cache_key = f"authors:stats:limit={limit}:offset={offset}"
# Функция для получения авторов из БД
async def fetch_authors_with_stats():
logger.debug(f"Выполняем запрос на получение авторов со статистикой: limit={limit}, offset={offset}, by={by}")
with local_session() as session:
# Базовый запрос для получения авторов
base_query = select(Author).where(Author.deleted_at.is_(None))
# Применяем сортировку
# vars for statistics sorting
stats_sort_field = None
stats_sort_direction = "desc"
if by:
if isinstance(by, dict):
logger.debug(f"Processing dict-based sorting: {by}")
# Обработка словаря параметров сортировки
from sqlalchemy import asc, desc, func
from orm.shout import ShoutAuthor
from orm.author import AuthorFollower
# Checking for order field in the dictionary
if "order" in by:
order_value = by["order"]
logger.debug(f"Found order field with value: {order_value}")
if order_value in ["shouts", "followers", "rating", "comments"]:
stats_sort_field = order_value
stats_sort_direction = "desc" # По умолчанию убывающая сортировка для статистики
logger.debug(f"Applying statistics-based sorting by: {stats_sort_field}")
elif order_value == "name":
# Sorting by name in ascending order
base_query = base_query.order_by(asc(Author.name))
logger.debug("Applying alphabetical sorting by name")
else:
# If order is not a stats field, treat it as a regular field
column = getattr(Author, order_value, None)
if column:
base_query = base_query.order_by(desc(column))
else:
# Regular sorting by fields
for field, direction in by.items():
column = getattr(Author, field, None)
if column:
if direction.lower() == "desc":
base_query = base_query.order_by(desc(column))
else:
base_query = base_query.order_by(column)
elif by == "new":
base_query = base_query.order_by(desc(Author.created_at))
elif by == "active":
base_query = base_query.order_by(desc(Author.last_seen))
else:
# По умолчанию сортируем по времени создания
base_query = base_query.order_by(desc(Author.created_at))
else:
base_query = base_query.order_by(desc(Author.created_at))
# If sorting by statistics, modify the query
if stats_sort_field == "shouts":
# Sorting by the number of shouts
from sqlalchemy import func, and_
from orm.shout import Shout, ShoutAuthor
subquery = (
select(
ShoutAuthor.author,
func.count(func.distinct(Shout.id)).label("shouts_count")
)
.select_from(ShoutAuthor)
.join(Shout, ShoutAuthor.shout == Shout.id)
.where(
and_(
Shout.deleted_at.is_(None),
Shout.published_at.is_not(None)
)
)
.group_by(ShoutAuthor.author)
.subquery()
)
base_query = (
base_query
.outerjoin(subquery, Author.id == subquery.c.author)
.order_by(desc(func.coalesce(subquery.c.shouts_count, 0)))
)
elif stats_sort_field == "followers":
# Sorting by the number of followers
from sqlalchemy import func
from orm.author import AuthorFollower
subquery = (
select(
AuthorFollower.author,
func.count(func.distinct(AuthorFollower.follower)).label("followers_count")
)
.select_from(AuthorFollower)
.group_by(AuthorFollower.author)
.subquery()
)
base_query = (
base_query
.outerjoin(subquery, Author.id == subquery.c.author)
.order_by(desc(func.coalesce(subquery.c.followers_count, 0)))
)
# Применяем лимит и смещение
base_query = base_query.limit(limit).offset(offset)
# Получаем авторов
authors = session.execute(base_query).scalars().all()
author_ids = [author.id for author in authors]
if not author_ids:
return []
# Оптимизированный запрос для получения статистики по публикациям для авторов
shouts_stats_query = f"""
SELECT sa.author, COUNT(DISTINCT s.id) as shouts_count
FROM shout_author sa
JOIN shout s ON sa.shout = s.id AND s.deleted_at IS NULL AND s.published_at IS NOT NULL
WHERE sa.author IN ({",".join(map(str, author_ids))})
GROUP BY sa.author
"""
shouts_stats = {row[0]: row[1] for row in session.execute(text(shouts_stats_query))}
# Запрос на получение статистики по подписчикам для авторов
followers_stats_query = f"""
SELECT author, COUNT(DISTINCT follower) as followers_count
FROM author_follower
WHERE author IN ({",".join(map(str, author_ids))})
GROUP BY author
"""
followers_stats = {row[0]: row[1] for row in session.execute(text(followers_stats_query))}
# Формируем результат с добавлением статистики
result = []
for author in authors:
author_dict = author.dict()
author_dict["stat"] = {
"shouts": shouts_stats.get(author.id, 0),
"followers": followers_stats.get(author.id, 0),
}
result.append(author_dict)
# Кешируем каждого автора отдельно для использования в других функциях
await cache_author(author_dict)
return result
# Используем универсальную функцию для кеширования запросов
return await cached_query(cache_key, fetch_authors_with_stats)
# Функция для инвалидации кеша авторов
async def invalidate_authors_cache(author_id=None):
"""
Инвалидирует кеши авторов при изменении данных.
Args:
author_id: Опциональный ID автора для точечной инвалидации.
Если не указан, инвалидируются все кеши авторов.
"""
if author_id:
# Точечная инвалидация конкретного автора
logger.debug(f"Инвалидация кеша для автора #{author_id}")
specific_keys = [
f"author:id:{author_id}",
f"author:followers:{author_id}",
f"author:follows-authors:{author_id}",
f"author:follows-topics:{author_id}",
f"author:follows-shouts:{author_id}",
]
# Получаем user_id автора, если есть
with local_session() as session:
author = session.query(Author).filter(Author.id == author_id).first()
if author and author.user:
specific_keys.append(f"author:user:{author.user.strip()}")
# Удаляем конкретные ключи
for key in specific_keys:
try:
await redis.execute("DEL", key)
logger.debug(f"Удален ключ кеша {key}")
except Exception as e:
logger.error(f"Ошибка при удалении ключа {key}: {e}")
# Также ищем и удаляем ключи коллекций, содержащих данные об этом авторе
collection_keys = await redis.execute("KEYS", "authors:stats:*")
if collection_keys:
await redis.execute("DEL", *collection_keys)
logger.debug(f"Удалено {len(collection_keys)} коллекционных ключей авторов")
else:
# Общая инвалидация всех кешей авторов
logger.debug("Полная инвалидация кеша авторов")
await invalidate_cache_by_prefix("authors")
@mutation.field("update_author")
@login_required
@@ -298,14 +51,10 @@ async def update_author(_, info, profile):
@query.field("get_authors_all")
async def get_authors_all(_, _info):
"""
Получает список всех авторов без статистики.
Returns:
list: Список всех авторов
"""
return await get_all_authors()
def get_authors_all(_, _info):
with local_session() as session:
authors = session.query(Author).all()
return authors
@query.field("get_author")
@@ -356,105 +105,145 @@ async def get_author_id(_, _info, user: str):
asyncio.create_task(cache_author(author_dict))
return author_with_stat
except Exception as exc:
logger.error(f"Error getting author: {exc}")
return None
import traceback
traceback.print_exc()
logger.error(exc)
@query.field("load_authors_by")
async def load_authors_by(_, _info, by, limit, offset):
"""
Загружает авторов по заданному критерию с пагинацией.
logger.debug(f"loading authors by {by}")
authors_query = select(Author)
Args:
by: Критерий сортировки авторов (new/active)
limit: Максимальное количество возвращаемых авторов
offset: Смещение для пагинации
if by.get("slug"):
authors_query = authors_query.filter(Author.slug.ilike(f"%{by['slug']}%"))
elif by.get("name"):
authors_query = authors_query.filter(Author.name.ilike(f"%{by['name']}%"))
elif by.get("topic"):
authors_query = (
authors_query.join(ShoutAuthor) # Первое соединение ShoutAuthor
.join(ShoutTopic, ShoutAuthor.shout == ShoutTopic.shout)
.join(Topic, ShoutTopic.topic == Topic.id)
.filter(Topic.slug == str(by["topic"]))
)
Returns:
list: Список авторов с учетом критерия
"""
# Используем оптимизированную функцию для получения авторов
return await get_authors_with_stats(limit, offset, by)
if by.get("last_seen"): # в unix time
before = int(time.time()) - by["last_seen"]
authors_query = authors_query.filter(Author.last_seen > before)
elif by.get("created_at"): # в unix time
before = int(time.time()) - by["created_at"]
authors_query = authors_query.filter(Author.created_at > before)
authors_query = authors_query.limit(limit).offset(offset)
with local_session() as session:
authors_nostat = session.execute(authors_query).all()
authors = []
for a in authors_nostat:
if isinstance(a, Author):
author_dict = await get_cached_author(a.id, get_with_stat)
if author_dict and isinstance(author_dict.get("shouts"), int):
authors.append(author_dict)
# order
order = by.get("order")
if order in ["shouts", "followers"]:
authors_query = authors_query.order_by(desc(text(f"{order}_stat")))
# group by
authors = get_with_stat(authors_query)
return authors or []
def get_author_id_from(slug="", user=None, author_id=None):
try:
author_id = None
if author_id:
return author_id
with local_session() as session:
author = None
if slug:
author = session.query(Author).filter(Author.slug == slug).first()
if author:
author_id = author.id
return author_id
if user:
author = session.query(Author).filter(Author.user == user).first()
if author:
author_id = author.id
except Exception as exc:
logger.error(exc)
if not slug and not user and not author_id:
raise ValueError("One of slug, user, or author_id must be provided")
author_query = select(Author.id)
if user:
author_query = author_query.filter(Author.user == user)
elif slug:
author_query = author_query.filter(Author.slug == slug)
elif author_id:
author_query = author_query.filter(Author.id == author_id)
with local_session() as session:
author_id_result = session.execute(author_query).first()
author_id = author_id_result[0] if author_id_result else None
if not author_id:
raise ValueError("Author not found")
return author_id
@query.field("get_author_follows")
async def get_author_follows(_, _info, slug="", user=None, author_id=0):
logger.debug(f"getting follows for @{slug}")
author_id = get_author_id_from(slug=slug, user=user, author_id=author_id)
if not author_id:
return {}
try:
author_id = get_author_id_from(slug, user, author_id)
followed_authors = await get_cached_follower_authors(author_id)
followed_topics = await get_cached_follower_topics(author_id)
if bool(author_id):
logger.debug(f"getting {author_id} follows authors")
authors = await get_cached_follower_authors(author_id)
topics = await get_cached_follower_topics(author_id)
return {
"topics": topics,
"authors": authors,
"communities": [{"id": 1, "name": "Дискурс", "slug": "discours", "pic": ""}],
}
except Exception:
import traceback
# TODO: Get followed communities too
return {
"authors": followed_authors,
"topics": followed_topics,
"communities": DEFAULT_COMMUNITIES,
"shouts": [],
}
traceback.print_exc()
return {"error": "Author not found"}
@query.field("get_author_follows_topics")
async def get_author_follows_topics(_, _info, slug="", user=None, author_id=None):
logger.debug(f"getting followed topics for @{slug}")
author_id = get_author_id_from(slug=slug, user=user, author_id=author_id)
if not author_id:
return []
followed_topics = await get_cached_follower_topics(author_id)
return followed_topics
try:
follower_id = get_author_id_from(slug, user, author_id)
topics = await get_cached_follower_topics(follower_id)
return topics
except Exception:
import traceback
traceback.print_exc()
@query.field("get_author_follows_authors")
async def get_author_follows_authors(_, _info, slug="", user=None, author_id=None):
logger.debug(f"getting followed authors for @{slug}")
author_id = get_author_id_from(slug=slug, user=user, author_id=author_id)
if not author_id:
return []
followed_authors = await get_cached_follower_authors(author_id)
return followed_authors
try:
follower_id = get_author_id_from(slug, user, author_id)
return await get_cached_follower_authors(follower_id)
except Exception:
import traceback
traceback.print_exc()
def create_author(user_id: str, slug: str, name: str = ""):
author = Author()
author.user = user_id # Связь с user_id из системы авторизации
author.slug = slug # Идентификатор из системы авторизации
author.created_at = author.updated_at = int(time.time())
author.name = name or slug # если не указано
with local_session() as session:
session.add(author)
session.commit()
return author
try:
author = None
if user_id:
author = session.query(Author).filter(Author.user == user_id).first()
elif slug:
author = session.query(Author).filter(Author.slug == slug).first()
if not author:
new_author = Author(user=user_id, slug=slug, name=name)
session.add(new_author)
session.commit()
logger.info(f"author created by webhook {new_author.dict()}")
except Exception as exc:
logger.debug(exc)
@query.field("get_author_followers")
async def get_author_followers(_, _info, slug: str = "", user: str = "", author_id: int = 0):
logger.debug(f"getting followers for author @{slug} or ID:{author_id}")
logger.debug(f"getting followers for @{slug}")
author_id = get_author_id_from(slug=slug, user=user, author_id=author_id)
if not author_id:
return []
followers = await get_cached_author_followers(author_id)
followers = []
if author_id:
followers = await get_cached_author_followers(author_id)
return followers

475
resolvers/draft.py
View File

@@ -1,475 +0,0 @@
import time
import trafilatura
from sqlalchemy.orm import joinedload
from cache.cache import (
cache_author,
cache_by_id,
cache_topic,
invalidate_shout_related_cache,
invalidate_shouts_cache,
)
from orm.author import Author
from orm.draft import Draft, DraftAuthor, DraftTopic
from orm.shout import Shout, ShoutAuthor, ShoutTopic
from orm.topic import Topic
from services.auth import login_required
from services.db import local_session
from services.notify import notify_shout
from services.schema import mutation, query
from services.search import search_service
from utils.html_wrapper import wrap_html_fragment
from utils.logger import root_logger as logger
def create_shout_from_draft(session, draft, author_id):
"""
Создаёт новый объект публикации (Shout) на основе черновика.
Args:
session: SQLAlchemy сессия (не используется, для совместимости)
draft (Draft): Объект черновика
author_id (int): ID автора публикации
Returns:
Shout: Новый объект публикации (не сохранённый в базе)
Пример:
>>> from orm.draft import Draft
>>> draft = Draft(id=1, title='Заголовок', body='Текст', slug='slug', created_by=1)
>>> shout = create_shout_from_draft(None, draft, 1)
>>> shout.title
'Заголовок'
>>> shout.body
'Текст'
>>> shout.created_by
1
"""
# Создаем новую публикацию
shout = Shout(
body=draft.body or "",
slug=draft.slug,
cover=draft.cover,
cover_caption=draft.cover_caption,
lead=draft.lead,
title=draft.title or "",
subtitle=draft.subtitle,
layout=draft.layout or "article",
media=draft.media or [],
lang=draft.lang or "ru",
seo=draft.seo,
created_by=author_id,
community=draft.community,
draft=draft.id,
deleted_at=None,
)
# Инициализируем пустые массивы для связей
shout.topics = []
shout.authors = []
return shout
@query.field("load_drafts")
@login_required
async def load_drafts(_, info):
"""
Загружает все черновики, доступные текущему пользователю.
Предварительно загружает связанные объекты (topics, authors, publication),
чтобы избежать ошибок с отсоединенными объектами при сериализации.
Returns:
dict: Список черновиков или сообщение об ошибке
"""
user_id = info.context.get("user_id")
author_dict = info.context.get("author", {})
author_id = author_dict.get("id")
if not user_id or not author_id:
return {"error": "User ID and author ID are required"}
try:
with local_session() as session:
# Предзагружаем authors, topics и связанную publication
drafts_query = (
session.query(Draft)
.options(
joinedload(Draft.topics),
joinedload(Draft.authors),
joinedload(Draft.publication) # Загружаем связанную публикацию
)
.filter(Draft.authors.any(Author.id == author_id))
)
drafts = drafts_query.all()
# Преобразуем объекты в словари, пока они в контексте сессии
drafts_data = []
for draft in drafts:
draft_dict = draft.dict()
# Всегда возвращаем массив для topics, даже если он пустой
draft_dict["topics"] = [topic.dict() for topic in (draft.topics or [])]
draft_dict["authors"] = [author.dict() for author in (draft.authors or [])]
# Добавляем информацию о публикации, если она есть
if draft.publication:
draft_dict["publication"] = {
"id": draft.publication.id,
"slug": draft.publication.slug,
"published_at": draft.publication.published_at
}
else:
draft_dict["publication"] = None
drafts_data.append(draft_dict)
return {"drafts": drafts_data}
except Exception as e:
logger.error(f"Failed to load drafts: {e}", exc_info=True)
return {"error": f"Failed to load drafts: {str(e)}"}
@mutation.field("create_draft")
@login_required
async def create_draft(_, info, draft_input):
"""Create a new draft.
Args:
info: GraphQL context
draft_input (dict): Draft data including optional fields:
- title (str, required) - заголовок черновика
- body (str, required) - текст черновика
- slug (str)
- etc.
Returns:
dict: Contains either:
- draft: The created draft object
- error: Error message if creation failed
Example:
>>> async def test_create():
... context = {'user_id': '123', 'author': {'id': 1}}
... info = type('Info', (), {'context': context})()
... result = await create_draft(None, info, {'title': 'Test'})
... assert result.get('error') is None
... assert result['draft'].title == 'Test'
... return result
"""
user_id = info.context.get("user_id")
author_dict = info.context.get("author", {})
author_id = author_dict.get("id")
if not user_id or not author_id:
return {"error": "Author ID is required"}
# Проверяем обязательные поля
if "body" not in draft_input or not draft_input["body"]:
draft_input["body"] = "" # Пустая строка вместо NULL
if "title" not in draft_input or not draft_input["title"]:
draft_input["title"] = "" # Пустая строка вместо NULL
# Проверяем slug - он должен быть или не пустым, или не передаваться вообще
if "slug" in draft_input and (draft_input["slug"] is None or draft_input["slug"] == ""):
# При создании черновика удаляем пустой slug из входных данных
del draft_input["slug"]
try:
with local_session() as session:
# Remove id from input if present since it's auto-generated
if "id" in draft_input:
del draft_input["id"]
# Добавляем текущее время создания и ID автора
draft_input["created_at"] = int(time.time())
draft_input["created_by"] = author_id
draft = Draft(**draft_input)
session.add(draft)
session.flush()
# Добавляем создателя как автора
da = DraftAuthor(shout=draft.id, author=author_id)
session.add(da)
session.commit()
return {"draft": draft}
except Exception as e:
logger.error(f"Failed to create draft: {e}", exc_info=True)
return {"error": f"Failed to create draft: {str(e)}"}
def generate_teaser(body, limit=300):
body_html = wrap_html_fragment(body)
body_text = trafilatura.extract(body_html, include_comments=False, include_tables=False)
body_teaser = ". ".join(body_text[:limit].split(". ")[:-1])
return body_teaser
@mutation.field("update_draft")
@login_required
async def update_draft(_, info, draft_id: int, draft_input):
"""Обновляет черновик публикации.
Args:
draft_id: ID черновика для обновления
draft_input: Данные для обновления черновика согласно схеме DraftInput:
- layout: String
- author_ids: [Int!]
- topic_ids: [Int!]
- main_topic_id: Int
- media: [MediaItemInput]
- lead: String
- subtitle: String
- lang: String
- seo: String
- body: String
- title: String
- slug: String
- cover: String
- cover_caption: String
Returns:
dict: Обновленный черновик или сообщение об ошибке
"""
user_id = info.context.get("user_id")
author_dict = info.context.get("author", {})
author_id = author_dict.get("id")
if not user_id or not author_id:
return {"error": "Author ID are required"}
try:
with local_session() as session:
draft = session.query(Draft).filter(Draft.id == draft_id).first()
if not draft:
return {"error": "Draft not found"}
# Фильтруем входные данные, оставляя только разрешенные поля
allowed_fields = {
"layout", "author_ids", "topic_ids", "main_topic_id",
"media", "lead", "subtitle", "lang", "seo", "body",
"title", "slug", "cover", "cover_caption"
}
filtered_input = {k: v for k, v in draft_input.items() if k in allowed_fields}
# Проверяем slug
if "slug" in filtered_input and not filtered_input["slug"]:
del filtered_input["slug"]
# Обновляем связи с авторами если переданы
if "author_ids" in filtered_input:
author_ids = filtered_input.pop("author_ids")
if author_ids:
# Очищаем текущие связи
session.query(DraftAuthor).filter(DraftAuthor.shout == draft_id).delete()
# Добавляем новые связи
for aid in author_ids:
da = DraftAuthor(shout=draft_id, author=aid)
session.add(da)
# Обновляем связи с темами если переданы
if "topic_ids" in filtered_input:
topic_ids = filtered_input.pop("topic_ids")
main_topic_id = filtered_input.pop("main_topic_id", None)
if topic_ids:
# Очищаем текущие связи
session.query(DraftTopic).filter(DraftTopic.shout == draft_id).delete()
# Добавляем новые связи
for tid in topic_ids:
dt = DraftTopic(
shout=draft_id,
topic=tid,
main=(tid == main_topic_id) if main_topic_id else False
)
session.add(dt)
# Генерируем SEO если не предоставлено
if "seo" not in filtered_input and not draft.seo:
body_src = filtered_input.get("body", draft.body)
lead_src = filtered_input.get("lead", draft.lead)
body_html = wrap_html_fragment(body_src)
lead_html = wrap_html_fragment(lead_src)
try:
body_text = trafilatura.extract(body_html, include_comments=False, include_tables=False) if body_src else None
lead_text = trafilatura.extract(lead_html, include_comments=False, include_tables=False) if lead_src else None
body_teaser = generate_teaser(body_text, 300) if body_text else ""
filtered_input["seo"] = lead_text if lead_text else body_teaser
except Exception as e:
logger.warning(f"Failed to generate SEO for draft {draft_id}: {e}")
# Обновляем основные поля черновика
for key, value in filtered_input.items():
setattr(draft, key, value)
# Обновляем метаданные
draft.updated_at = int(time.time())
draft.updated_by = author_id
session.commit()
# Преобразуем объект в словарь для ответа
draft_dict = draft.dict()
draft_dict["topics"] = [topic.dict() for topic in draft.topics]
draft_dict["authors"] = [author.dict() for author in draft.authors]
# Добавляем объект автора в updated_by
draft_dict["updated_by"] = author_dict
return {"draft": draft_dict}
except Exception as e:
logger.error(f"Failed to update draft: {e}", exc_info=True)
return {"error": f"Failed to update draft: {str(e)}"}
@mutation.field("delete_draft")
@login_required
async def delete_draft(_, info, draft_id: int):
author_dict = info.context.get("author", {})
author_id = author_dict.get("id")
with local_session() as session:
draft = session.query(Draft).filter(Draft.id == draft_id).first()
if not draft:
return {"error": "Draft not found"}
if author_id != draft.created_by and draft.authors.filter(Author.id == author_id).count() == 0:
return {"error": "You are not allowed to delete this draft"}
session.delete(draft)
session.commit()
return {"draft": draft}
def validate_html_content(html_content: str) -> tuple[bool, str]:
"""
Проверяет валидность HTML контента через trafilatura.
Args:
html_content: HTML строка для проверки
Returns:
tuple[bool, str]: (валидность, сообщение об ошибке)
Example:
>>> is_valid, error = validate_html_content("<p>Valid HTML</p>")
>>> is_valid
True
>>> error
''
>>> is_valid, error = validate_html_content("Invalid < HTML")
>>> is_valid
False
>>> 'Invalid HTML' in error
True
"""
if not html_content or not html_content.strip():
return False, "Content is empty"
try:
html_content = wrap_html_fragment(html_content)
extracted = trafilatura.extract(html_content)
if not extracted:
return False, "Invalid HTML structure or empty content"
return True, ""
except Exception as e:
logger.error(f"HTML validation error: {e}", exc_info=True)
return False, f"Invalid HTML content: {str(e)}"
@mutation.field("publish_draft")
@login_required
async def publish_draft(_, info, draft_id: int):
"""
Публикует черновик, создавая новый Shout или обновляя существующий.
Args:
draft_id (int): ID черновика для публикации
Returns:
dict: Результат публикации с shout или сообщением об ошибке
"""
user_id = info.context.get("user_id")
author_dict = info.context.get("author", {})
author_id = author_dict.get("id")
if not user_id or not author_id:
return {"error": "Author ID is required"}
try:
with local_session() as session:
# Загружаем черновик со всеми связями
draft = (
session.query(Draft)
.options(
joinedload(Draft.topics),
joinedload(Draft.authors),
joinedload(Draft.publication)
)
.filter(Draft.id == draft_id)
.first()
)
if not draft:
return {"error": "Draft not found"}
# Проверка валидности HTML в body
is_valid, error = validate_html_content(draft.body)
if not is_valid:
return {"error": f"Cannot publish draft: {error}"}
# Проверяем, есть ли уже публикация для этого черновика
if draft.publication:
shout = draft.publication
# Обновляем существующую публикацию
for field in ["body", "title", "subtitle", "lead", "cover", "cover_caption", "media", "lang", "seo"]:
if hasattr(draft, field):
setattr(shout, field, getattr(draft, field))
shout.updated_at = int(time.time())
shout.updated_by = author_id
else:
# Создаем новую публикацию
shout = create_shout_from_draft(session, draft, author_id)
now = int(time.time())
shout.created_at = now
shout.published_at = now
session.add(shout)
session.flush() # Получаем ID нового шаута
# Очищаем существующие связи
session.query(ShoutAuthor).filter(ShoutAuthor.shout == shout.id).delete()
session.query(ShoutTopic).filter(ShoutTopic.shout == shout.id).delete()
# Добавляем авторов
for author in (draft.authors or []):
sa = ShoutAuthor(shout=shout.id, author=author.id)
session.add(sa)
# Добавляем темы
for topic in (draft.topics or []):
st = ShoutTopic(
topic=topic.id,
shout=shout.id,
main=topic.main if hasattr(topic, "main") else False
)
session.add(st)
session.commit()
# Инвалидируем кеш
invalidate_shouts_cache()
invalidate_shout_related_cache(shout.id)
# Уведомляем о публикации
await notify_shout(shout.id)
# Обновляем поисковый индекс
search_service.index_shout(shout)
logger.info(f"Successfully published shout #{shout.id} from draft #{draft_id}")
logger.debug(f"Shout data: {shout.dict()}")
return {"shout": shout}
except Exception as e:
logger.error(f"Failed to publish draft {draft_id}: {e}", exc_info=True)
return {"error": f"Failed to publish draft: {str(e)}"}

715
resolvers/editor.py
View File

@@ -1,19 +1,11 @@
import time
import orjson
import trafilatura
from sqlalchemy import and_, desc, select
from sqlalchemy.orm import joinedload, selectinload
from sqlalchemy.orm import joinedload
from sqlalchemy.sql.functions import coalesce
from cache.cache import (
cache_author,
cache_topic,
invalidate_shout_related_cache,
invalidate_shouts_cache,
)
from cache.cache import cache_author, cache_topic
from orm.author import Author
from orm.draft import Draft
from orm.shout import Shout, ShoutAuthor, ShoutTopic
from orm.topic import Topic
from resolvers.follower import follow, unfollow
@@ -23,28 +15,10 @@ from services.db import local_session
from services.notify import notify_shout
from services.schema import mutation, query
from services.search import search_service
from utils.html_wrapper import wrap_html_fragment
from utils.logger import root_logger as logger
async def cache_by_id(entity, entity_id: int, cache_method):
"""Cache an entity by its ID using the provided cache method.
Args:
entity: The SQLAlchemy model class to query
entity_id (int): The ID of the entity to cache
cache_method: The caching function to use
Returns:
dict: The cached entity data if successful, None if entity not found
Example:
>>> async def test_cache():
... author = await cache_by_id(Author, 1, cache_author)
... assert author['id'] == 1
... assert 'name' in author
... return author
"""
caching_query = select(entity).filter(entity.id == entity_id)
result = get_with_stat(caching_query)
if not result or not result[0]:
@@ -59,34 +33,7 @@ async def cache_by_id(entity, entity_id: int, cache_method):
@query.field("get_my_shout")
@login_required
async def get_my_shout(_, info, shout_id: int):
"""Get a shout by ID if the requesting user has permission to view it.
DEPRECATED: use `load_drafts` instead
Args:
info: GraphQL resolver info containing context
shout_id (int): ID of the shout to retrieve
Returns:
dict: Contains either:
- error (str): Error message if retrieval failed
- shout (Shout): The requested shout if found and accessible
Permissions:
User must be:
- The shout creator
- Listed as an author
- Have editor role
Example:
>>> async def test_get_my_shout():
... context = {'user_id': '123', 'author': {'id': 1}, 'roles': []}
... info = type('Info', (), {'context': context})()
... result = await get_my_shout(None, info, 1)
... assert result['error'] is None
... assert result['shout'].id == 1
... return result
"""
# logger.debug(info)
user_id = info.context.get("user_id", "")
author_dict = info.context.get("author", {})
author_id = author_dict.get("id")
@@ -105,26 +52,13 @@ async def get_my_shout(_, info, shout_id: int):
if not shout:
return {"error": "no shout found", "shout": None}
# Преобразуем media JSON в список объектов MediaItem
if hasattr(shout, "media") and shout.media:
if isinstance(shout.media, str):
try:
shout.media = orjson.loads(shout.media)
except Exception as e:
logger.error(f"Error parsing shout media: {e}")
shout.media = []
if not isinstance(shout.media, list):
shout.media = [shout.media] if shout.media else []
else:
shout.media = []
logger.debug(f"got {len(shout.authors)} shout authors, created by {shout.created_by}")
is_editor = "editor" in roles
logger.debug(f"viewer is{'' if is_editor else ' not'} editor")
logger.debug(f'viewer is{'' if is_editor else ' not'} editor')
is_creator = author_id == shout.created_by
logger.debug(f"viewer is{'' if is_creator else ' not'} creator")
logger.debug(f'viewer is{'' if is_creator else ' not'} creator')
is_author = bool(list(filter(lambda x: x.id == int(author_id), [x for x in shout.authors])))
logger.debug(f"viewer is{'' if is_creator else ' not'} author")
logger.debug(f'viewer is{'' if is_creator else ' not'} author')
can_edit = is_editor or is_author or is_creator
if not can_edit:
@@ -157,284 +91,155 @@ async def get_shouts_drafts(_, info):
return {"shouts": shouts}
# @mutation.field("create_shout")
# @login_required
@mutation.field("create_shout")
@login_required
async def create_shout(_, info, inp):
logger.info(f"Starting create_shout with input: {inp}")
user_id = info.context.get("user_id")
author_dict = info.context.get("author")
logger.debug(f"Context user_id: {user_id}, author: {author_dict}")
if not author_dict:
logger.error("Author profile not found in context")
return {"error": "author profile was not found"}
author_id = author_dict.get("id")
if user_id and author_id:
try:
with local_session() as session:
author_id = int(author_id)
current_time = int(time.time())
slug = inp.get("slug") or f"draft-{current_time}"
with local_session() as session:
author_id = int(author_id)
current_time = int(time.time())
slug = inp.get("slug") or f"draft-{current_time}"
shout_dict = {
"title": inp.get("title", ""),
"subtitle": inp.get("subtitle", ""),
"lead": inp.get("lead", ""),
"description": inp.get("description", ""),
"body": inp.get("body", ""),
"layout": inp.get("layout", "article"),
"created_by": author_id,
"authors": [],
"slug": slug,
"topics": inp.get("topics", []),
"published_at": None,
"community": 1,
"created_at": current_time, # Set created_at as Unix timestamp
}
same_slug_shout = session.query(Shout).filter(Shout.slug == shout_dict.get("slug")).first()
c = 1
while same_slug_shout is not None:
same_slug_shout = session.query(Shout).filter(Shout.slug == shout_dict.get("slug")).first()
c += 1
shout_dict["slug"] += f"-{c}"
new_shout = Shout(**shout_dict)
session.add(new_shout)
session.commit()
logger.info(f"Creating shout with input: {inp}")
# Создаем публикацию без topics
body = inp.get("body", "")
lead = inp.get("lead", "")
body_html = wrap_html_fragment(body)
lead_html = wrap_html_fragment(lead)
body_text = trafilatura.extract(body_html)
lead_text = trafilatura.extract(lead_html)
seo = inp.get("seo", lead_text.strip() or body_text.strip()[:300].split(". ")[:-1].join(". "))
new_shout = Shout(
slug=slug,
body=body,
seo=seo,
lead=lead,
layout=inp.get("layout", "article"),
title=inp.get("title", ""),
created_by=author_id,
created_at=current_time,
community=1,
)
# Проверяем уникальность slug
logger.debug(f"Checking for existing slug: {slug}")
same_slug_shout = session.query(Shout).filter(Shout.slug == new_shout.slug).first()
c = 1
while same_slug_shout is not None:
logger.debug(f"Found duplicate slug, trying iteration {c}")
new_shout.slug = f"{slug}-{c}"
same_slug_shout = session.query(Shout).filter(Shout.slug == new_shout.slug).first()
c += 1
try:
logger.info("Creating new shout object")
session.add(new_shout)
session.commit()
logger.info(f"Created shout with ID: {new_shout.id}")
except Exception as e:
logger.error(f"Error creating shout object: {e}", exc_info=True)
return {"error": f"Database error: {str(e)}"}
# Связываем с автором
try:
logger.debug(f"Linking author {author_id} to shout {new_shout.id}")
sa = ShoutAuthor(shout=new_shout.id, author=author_id)
# NOTE: requesting new shout back
shout = session.query(Shout).where(Shout.slug == slug).first()
if shout:
# Проверка на существование записи
existing_sa = session.query(ShoutAuthor).filter_by(shout=shout.id, author=author_id).first()
if not existing_sa:
sa = ShoutAuthor(shout=shout.id, author=author_id)
session.add(sa)
except Exception as e:
logger.error(f"Error linking author: {e}", exc_info=True)
return {"error": f"Error linking author: {str(e)}"}
# Связываем с темами
topics = session.query(Topic).filter(Topic.slug.in_(inp.get("topics", []))).all()
for topic in topics:
existing_st = session.query(ShoutTopic).filter_by(shout=shout.id, author=topic.id).first()
if not existing_st:
t = ShoutTopic(topic=topic.id, shout=shout.id)
session.add(t)
input_topics = inp.get("topics", [])
if input_topics:
try:
logger.debug(f"Linking topics: {[t.slug for t in input_topics]}")
main_topic = inp.get("main_topic")
for topic in input_topics:
st = ShoutTopic(
topic=topic.id,
shout=new_shout.id,
main=(topic.slug == main_topic) if main_topic else False,
)
session.add(st)
logger.debug(f"Added topic {topic.slug} {'(main)' if st.main else ''}")
except Exception as e:
logger.error(f"Error linking topics: {e}", exc_info=True)
return {"error": f"Error linking topics: {str(e)}"}
session.commit()
try:
session.commit()
logger.info("Final commit successful")
except Exception as e:
logger.error(f"Error in final commit: {e}", exc_info=True)
return {"error": f"Error in final commit: {str(e)}"}
follow(None, info, "shout", shout.slug)
# Получаем созданную публикацию
shout = session.query(Shout).filter(Shout.id == new_shout.id).first()
# notifier
# await notify_shout(shout_dict, 'create')
# Подписываем автора
try:
logger.debug("Following created shout")
await follow(None, info, "shout", shout.slug)
except Exception as e:
logger.warning(f"Error following shout: {e}", exc_info=True)
logger.info(f"Successfully created shout {shout.id}")
return {"shout": shout}
except Exception as e:
logger.error(f"Unexpected error in create_shout: {e}", exc_info=True)
return {"error": f"Unexpected error: {str(e)}"}
error_msg = "cant create shout" if user_id else "unauthorized"
logger.error(f"Create shout failed: {error_msg}")
return {"error": error_msg}
return {"error": "cant create shout" if user_id else "unauthorized"}
def patch_main_topic(session, main_topic_slug, shout):
"""Update the main topic for a shout."""
logger.info(f"Starting patch_main_topic for shout#{shout.id} with slug '{main_topic_slug}'")
logger.debug(f"Current shout topics: {[(t.topic.slug, t.main) for t in shout.topics]}")
def patch_main_topic(session, main_topic, shout):
with session.begin():
# Получаем текущий главный топик
old_main = (
shout = session.query(Shout).options(joinedload(Shout.topics)).filter(Shout.id == shout.id).first()
if not shout:
return
old_main_topic = (
session.query(ShoutTopic).filter(and_(ShoutTopic.shout == shout.id, ShoutTopic.main.is_(True))).first()
)
if old_main:
logger.info(f"Found current main topic: {old_main.topic.slug}")
else:
logger.info("No current main topic found")
# Находим новый главный топик
main_topic = session.query(Topic).filter(Topic.slug == main_topic_slug).first()
if not main_topic:
logger.error(f"Main topic with slug '{main_topic_slug}' not found")
return
main_topic = session.query(Topic).filter(Topic.slug == main_topic).first()
logger.info(f"Found new main topic: {main_topic.slug} (id={main_topic.id})")
if main_topic:
new_main_topic = (
session.query(ShoutTopic)
.filter(and_(ShoutTopic.shout == shout.id, ShoutTopic.topic == main_topic.id))
.first()
)
# Находим связь с новым главным топиком
new_main = (
session.query(ShoutTopic)
.filter(and_(ShoutTopic.shout == shout.id, ShoutTopic.topic == main_topic.id))
.first()
)
logger.debug(f"Found new main topic relation: {new_main is not None}")
if old_main_topic and new_main_topic and old_main_topic is not new_main_topic:
ShoutTopic.update(old_main_topic, {"main": False})
session.add(old_main_topic)
if old_main and new_main and old_main is not new_main:
logger.info(f"Updating main topic flags: {old_main.topic.slug} -> {new_main.topic.slug}")
old_main.main = False
session.add(old_main)
new_main.main = True
session.add(new_main)
session.flush()
logger.info(f"Main topic updated for shout#{shout.id}")
else:
logger.warning(f"No changes needed for main topic (old={old_main is not None}, new={new_main is not None})")
ShoutTopic.update(new_main_topic, {"main": True})
session.add(new_main_topic)
def patch_topics(session, shout, topics_input):
"""Update the topics associated with a shout.
Args:
session: SQLAlchemy session
shout (Shout): The shout to update
topics_input (list): List of topic dicts with fields:
- id (int): Topic ID (<0 for new topics)
- slug (str): Topic slug
- title (str): Topic title (for new topics)
Side Effects:
- Creates new topics if needed
- Updates shout-topic associations
- Refreshes shout object with new topics
Example:
>>> def test_patch_topics():
... topics = [
... {'id': -1, 'slug': 'new-topic', 'title': 'New Topic'},
... {'id': 1, 'slug': 'existing-topic'}
... ]
... with local_session() as session:
... shout = session.query(Shout).first()
... patch_topics(session, shout, topics)
... assert len(shout.topics) == 2
... assert any(t.slug == 'new-topic' for t in shout.topics)
... return shout.topics
"""
logger.info(f"Starting patch_topics for shout#{shout.id}")
logger.info(f"Received topics_input: {topics_input}")
# Создаем новые топики если есть
new_topics_to_link = [Topic(**new_topic) for new_topic in topics_input if new_topic["id"] < 0]
if new_topics_to_link:
logger.info(f"Creating new topics: {[t.dict() for t in new_topics_to_link]}")
session.add_all(new_topics_to_link)
session.flush()
session.commit()
# Получаем текущие связи
current_links = session.query(ShoutTopic).filter(ShoutTopic.shout == shout.id).all()
logger.info(f"Current topic links: {[{t.topic: t.main} for t in current_links]}")
for new_topic_to_link in new_topics_to_link:
created_unlinked_topic = ShoutTopic(shout=shout.id, topic=new_topic_to_link.id)
session.add(created_unlinked_topic)
# Удаляем старые связи
if current_links:
logger.info(f"Removing old topic links for shout#{shout.id}")
for link in current_links:
session.delete(link)
session.flush()
existing_topics_input = [topic_input for topic_input in topics_input if topic_input.get("id", 0) > 0]
existing_topic_to_link_ids = [
existing_topic_input["id"]
for existing_topic_input in existing_topics_input
if existing_topic_input["id"] not in [topic.id for topic in shout.topics]
]
# Создаем новые связи
for topic_input in topics_input:
topic_id = topic_input["id"]
if topic_id < 0:
topic = next(t for t in new_topics_to_link if t.slug == topic_input["slug"])
topic_id = topic.id
for existing_topic_to_link_id in existing_topic_to_link_ids:
created_unlinked_topic = ShoutTopic(shout=shout.id, topic=existing_topic_to_link_id)
session.add(created_unlinked_topic)
logger.info(f"Creating new topic link: shout#{shout.id} -> topic#{topic_id}")
new_link = ShoutTopic(shout=shout.id, topic=topic_id, main=False)
session.add(new_link)
topic_to_unlink_ids = [
topic.id
for topic in shout.topics
if topic.id not in [topic_input["id"] for topic_input in existing_topics_input]
]
session.flush()
# Обновляем связи в объекте шаута
session.refresh(shout)
logger.info(f"Successfully updated topics for shout#{shout.id}")
logger.info(f"Final shout topics: {[t.dict() for t in shout.topics]}")
session.query(ShoutTopic).filter(
and_(ShoutTopic.shout == shout.id, ShoutTopic.topic.in_(topic_to_unlink_ids))
).delete(synchronize_session=False)
# @mutation.field("update_shout")
# @login_required
@mutation.field("update_shout")
@login_required
async def update_shout(_, info, shout_id: int, shout_input=None, publish=False):
logger.info(f"Starting update_shout with id={shout_id}, publish={publish}")
logger.debug(f"Full shout_input: {shout_input}") # DraftInput
user_id = info.context.get("user_id")
roles = info.context.get("roles", [])
author_dict = info.context.get("author")
if not author_dict:
logger.error("Author profile not found")
return {"error": "author profile was not found"}
author_id = author_dict.get("id")
shout_input = shout_input or {}
current_time = int(time.time())
shout_id = shout_id or shout_input.get("id", shout_id)
slug = shout_input.get("slug")
if not user_id:
logger.error("Unauthorized update attempt")
return {"error": "unauthorized"}
try:
with local_session() as session:
if author_id:
logger.info(f"Processing update for shout#{shout_id} by author #{author_id}")
shout_by_id = (
session.query(Shout)
.options(joinedload(Shout.topics).joinedload(ShoutTopic.topic), joinedload(Shout.authors))
.filter(Shout.id == shout_id)
.first()
)
logger.info(f"author for shout#{shout_id} detected author #{author_id}")
shout_by_id = session.query(Shout).filter(Shout.id == shout_id).first()
if not shout_by_id:
logger.error(f"shout#{shout_id} not found")
return {"error": "shout not found"}
logger.info(f"Found shout#{shout_id}")
# Логируем текущие топики
current_topics = (
[{"id": t.id, "slug": t.slug, "title": t.title} for t in shout_by_id.topics]
if shout_by_id.topics
else []
)
logger.info(f"Current topics for shout#{shout_id}: {current_topics}")
logger.info(f"shout#{shout_id} found")
if slug != shout_by_id.slug:
same_slug_shout = session.query(Shout).filter(Shout.slug == slug).first()
@@ -447,168 +252,55 @@ async def update_shout(_, info, shout_id: int, shout_input=None, publish=False):
logger.info(f"shout#{shout_id} slug patched")
if filter(lambda x: x.id == author_id, [x for x in shout_by_id.authors]) or "editor" in roles:
logger.info(f"Author #{author_id} has permission to edit shout#{shout_id}")
logger.info(f"shout#{shout_id} is author or editor")
# topics patch
topics_input = shout_input.get("topics")
if topics_input:
logger.info(f"Received topics_input for shout#{shout_id}: {topics_input}")
try:
patch_topics(session, shout_by_id, topics_input)
logger.info(f"Successfully patched topics for shout#{shout_id}")
# Обновляем связи в сессии после patch_topics
session.refresh(shout_by_id)
except Exception as e:
logger.error(f"Error patching topics: {e}", exc_info=True)
return {"error": f"Failed to update topics: {str(e)}"}
logger.info(f"topics_input: {topics_input}")
patch_topics(session, shout_by_id, topics_input)
del shout_input["topics"]
for tpc in topics_input:
await cache_by_id(Topic, tpc["id"], cache_topic)
else:
logger.warning(f"No topics_input received for shout#{shout_id}")
# main topic
main_topic = shout_input.get("main_topic")
if main_topic:
logger.info(f"Updating main topic for shout#{shout_id} to {main_topic}")
patch_main_topic(session, main_topic, shout_by_id)
shout_input["updated_at"] = current_time
if publish:
logger.info(f"Publishing shout#{shout_id}")
shout_input["published_at"] = current_time
# Проверяем наличие связи с автором
logger.info(f"Checking author link for shout#{shout_id} and author#{author_id}")
author_link = (
session.query(ShoutAuthor)
.filter(and_(ShoutAuthor.shout == shout_id, ShoutAuthor.author == author_id))
.first()
)
if not author_link:
logger.info(f"Adding missing author link for shout#{shout_id}")
sa = ShoutAuthor(shout=shout_id, author=author_id)
session.add(sa)
session.flush()
logger.info("Author link added successfully")
else:
logger.info("Author link already exists")
# Логируем финальное состояние перед сохранением
logger.info(f"Final shout_input for update: {shout_input}")
shout_input["published_at"] = current_time if publish else None
Shout.update(shout_by_id, shout_input)
session.add(shout_by_id)
session.commit()
try:
session.commit()
# Обновляем объект после коммита чтобы получить все связи
session.refresh(shout_by_id)
logger.info(f"Successfully committed updates for shout#{shout_id}")
except Exception as e:
logger.error(f"Commit failed: {e}", exc_info=True)
return {"error": f"Failed to save changes: {str(e)}"}
# После обновления проверяем топики
updated_topics = (
[{"id": t.id, "slug": t.slug, "title": t.title} for t in shout_by_id.topics]
if shout_by_id.topics
else []
)
logger.info(f"Updated topics for shout#{shout_id}: {updated_topics}")
# Инвалидация кэша после обновления
try:
logger.info("Invalidating cache after shout update")
cache_keys = [
"feed", # лента
f"author_{author_id}", # публикации автора
"random_top", # случайные топовые
"unrated", # неоцененные
]
# Добавляем ключи для тем публикации
for topic in shout_by_id.topics:
cache_keys.append(f"topic_{topic.id}")
cache_keys.append(f"topic_shouts_{topic.id}")
await invalidate_shouts_cache(cache_keys)
await invalidate_shout_related_cache(shout_by_id, author_id)
# Обновляем кэш тем и авторов
for topic in shout_by_id.topics:
await cache_by_id(Topic, topic.id, cache_topic)
for author in shout_by_id.authors:
await cache_author(author.dict())
logger.info("Cache invalidated successfully")
except Exception as cache_error:
logger.warning(f"Cache invalidation error: {cache_error}", exc_info=True)
shout_dict = shout_by_id.dict()
if not publish:
await notify_shout(shout_by_id.dict(), "update")
await notify_shout(shout_dict, "update")
else:
await notify_shout(shout_by_id.dict(), "published")
await notify_shout(shout_dict, "published")
# search service indexing
search_service.index(shout_by_id)
for a in shout_by_id.authors:
await cache_by_id(Author, a.id, cache_author)
logger.info(f"shout#{shout_id} updated")
# Получаем полные данные шаута со связями
shout_with_relations = (
session.query(Shout)
.options(joinedload(Shout.topics).joinedload(ShoutTopic.topic), joinedload(Shout.authors))
.filter(Shout.id == shout_id)
.first()
)
# Создаем словарь с базовыми полями
shout_dict = shout_with_relations.dict()
# Явно добавляем связанные данные
shout_dict["topics"] = (
[
{"id": topic.id, "slug": topic.slug, "title": topic.title}
for topic in shout_with_relations.topics
]
if shout_with_relations.topics
else []
)
# Add main_topic to the shout dictionary
shout_dict["main_topic"] = get_main_topic(shout_with_relations.topics)
shout_dict["authors"] = (
[
{"id": author.id, "name": author.name, "slug": author.slug}
for author in shout_with_relations.authors
]
if shout_with_relations.authors
else []
)
logger.info(f"Final shout data with relations: {shout_dict}")
logger.debug(
f"Loaded topics details: {[(t.topic.slug if t.topic else 'no-topic', t.main) for t in shout_with_relations.topics]}"
)
return {"shout": shout_dict, "error": None}
else:
logger.warning(f"Access denied: author #{author_id} cannot edit shout#{shout_id}")
logger.warning(f"shout#{shout_id} is not author or editor")
return {"error": "access denied", "shout": None}
except Exception as exc:
logger.error(f"Unexpected error in update_shout: {exc}", exc_info=True)
logger.error(f"Failed input data: {shout_input}")
return {"error": "cant update shout"}
import traceback
traceback.print_exc()
logger.error(exc)
logger.error(f" cannot update with data: {shout_input}")
return {"error": "cant update shout"}
# @mutation.field("delete_shout")
# @login_required
@mutation.field("delete_shout")
@login_required
async def delete_shout(_, info, shout_id: int):
user_id = info.context.get("user_id")
roles = info.context.get("roles", [])
@@ -643,184 +335,3 @@ async def delete_shout(_, info, shout_id: int):
return {"error": None}
else:
return {"error": "access denied"}
def get_main_topic(topics):
"""Get the main topic from a list of ShoutTopic objects."""
logger.info(f"Starting get_main_topic with {len(topics) if topics else 0} topics")
logger.debug(
f"Topics data: {[(t.slug, getattr(t, 'main', False)) for t in topics] if topics else []}"
)
if not topics:
logger.warning("No topics provided to get_main_topic")
return {"id": 0, "title": "no topic", "slug": "notopic", "is_main": True}
# Проверяем, является ли topics списком объектов ShoutTopic или Topic
if hasattr(topics[0], 'topic') and topics[0].topic:
# Для ShoutTopic объектов (старый формат)
# Find first main topic in original order
main_topic_rel = next((st for st in topics if getattr(st, 'main', False)), None)
logger.debug(
f"Found main topic relation: {main_topic_rel.topic.slug if main_topic_rel and main_topic_rel.topic else None}"
)
if main_topic_rel and main_topic_rel.topic:
result = {
"slug": main_topic_rel.topic.slug,
"title": main_topic_rel.topic.title,
"id": main_topic_rel.topic.id,
"is_main": True,
}
logger.info(f"Returning main topic: {result}")
return result
# If no main found but topics exist, return first
if topics and topics[0].topic:
logger.info(f"No main topic found, using first topic: {topics[0].topic.slug}")
result = {
"slug": topics[0].topic.slug,
"title": topics[0].topic.title,
"id": topics[0].topic.id,
"is_main": True,
}
return result
else:
# Для Topic объектов (новый формат из selectinload)
# После смены на selectinload у нас просто список Topic объектов
if topics:
logger.info(f"Using first topic as main: {topics[0].slug}")
result = {
"slug": topics[0].slug,
"title": topics[0].title,
"id": topics[0].id,
"is_main": True,
}
return result
logger.warning("No valid topics found, returning default")
return {"slug": "notopic", "title": "no topic", "id": 0, "is_main": True}
@mutation.field("unpublish_shout")
@login_required
async def unpublish_shout(_, info, shout_id: int):
"""Снимает публикацию (shout) с публикации.
Предзагружает связанный черновик (draft) и его авторов/темы, чтобы избежать
ошибок при последующем доступе к ним в GraphQL.
Args:
shout_id: ID публикации для снятия с публикации
Returns:
dict: Снятая с публикации публикация или сообщение об ошибке
"""
author_dict = info.context.get("author", {})
author_id = author_dict.get("id")
if not author_id:
# В идеале нужна проверка прав, имеет ли автор право снимать публикацию
return {"error": "Author ID is required"}
shout = None
with local_session() as session:
try:
# Загружаем Shout со всеми связями для правильного формирования ответа
shout = (
session.query(Shout)
.options(
joinedload(Shout.authors),
selectinload(Shout.topics)
)
.filter(Shout.id == shout_id)
.first()
)
if not shout:
logger.warning(f"Shout not found for unpublish: ID {shout_id}")
return {"error": "Shout not found"}
# Если у публикации есть связанный черновик, загружаем его с relationships
if shout.draft:
# Отдельно загружаем черновик с его связями
draft = (
session.query(Draft)
.options(
selectinload(Draft.authors),
selectinload(Draft.topics)
)
.filter(Draft.id == shout.draft)
.first()
)
# Связываем черновик с публикацией вручную для доступа через API
if draft:
shout.draft_obj = draft
# TODO: Добавить проверку прав доступа, если необходимо
# if author_id not in [a.id for a in shout.authors]: # Требует selectinload(Shout.authors) выше
# logger.warning(f"Author {author_id} denied unpublishing shout {shout_id}")
# return {"error": "Access denied"}
# Запоминаем старый slug и id для формирования поля publication
shout_slug = shout.slug
shout_id_for_publication = shout.id
# Снимаем с публикации (устанавливаем published_at в None)
shout.published_at = None
session.commit()
# Формируем полноценный словарь для ответа
shout_dict = shout.dict()
# Добавляем связанные данные
shout_dict["topics"] = (
[
{"id": topic.id, "slug": topic.slug, "title": topic.title}
for topic in shout.topics
]
if shout.topics
else []
)
# Добавляем main_topic
shout_dict["main_topic"] = get_main_topic(shout.topics)
# Добавляем авторов
shout_dict["authors"] = (
[
{"id": author.id, "name": author.name, "slug": author.slug}
for author in shout.authors
]
if shout.authors
else []
)
# Важно! Обновляем поле publication, отражая состояние "снят с публикации"
shout_dict["publication"] = {
"id": shout_id_for_publication,
"slug": shout_slug,
"published_at": None # Ключевое изменение - устанавливаем published_at в None
}
# Инвалидация кэша
try:
cache_keys = [
"feed", # лента
f"author_{author_id}", # публикации автора
"random_top", # случайные топовые
"unrated", # неоцененные
]
await invalidate_shout_related_cache(shout, author_id)
await invalidate_shouts_cache(cache_keys)
logger.info(f"Cache invalidated after unpublishing shout {shout_id}")
except Exception as cache_err:
logger.error(f"Failed to invalidate cache for unpublish shout {shout_id}: {cache_err}")
except Exception as e:
session.rollback()
logger.error(f"Failed to unpublish shout {shout_id}: {e}", exc_info=True)
return {"error": f"Failed to unpublish shout: {str(e)}"}
# Возвращаем сформированный словарь вместо объекта
logger.info(f"Shout {shout_id} unpublished successfully by author {author_id}")
return {"shout": shout_dict}

25
resolvers/feed.py
View File

@@ -5,12 +5,7 @@ from sqlalchemy import and_, select
from orm.author import Author, AuthorFollower
from orm.shout import Shout, ShoutAuthor, ShoutReactionsFollower, ShoutTopic
from orm.topic import Topic, TopicFollower
from resolvers.reader import (
apply_options,
get_shouts_with_links,
has_field,
query_with_stat,
)
from resolvers.reader import apply_options, get_shouts_with_links, has_field, query_with_stat
from services.auth import login_required
from services.db import local_session
from services.schema import query
@@ -178,21 +173,3 @@ async def load_shouts_with_topic(_, info, slug: str, options) -> List[Shout]:
except Exception as error:
logger.debug(error)
return []
def apply_filters(q, filters):
"""
Применяет фильтры к запросу
"""
logger.info(f"Applying filters: {filters}")
if filters.get("published"):
q = q.filter(Shout.published_at.is_not(None))
logger.info("Added published filter")
if filters.get("topic"):
topic_slug = filters["topic"]
q = q.join(ShoutTopic).join(Topic).filter(Topic.slug == topic_slug)
logger.info(f"Added topic filter: {topic_slug}")
return q

10
resolvers/notifier.py
View File

@@ -1,7 +1,7 @@
import json
import time
from typing import List, Tuple
import orjson
from sqlalchemy import and_, select
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import aliased
@@ -115,7 +115,7 @@ def get_notifications_grouped(author_id: int, after: int = 0, limit: int = 10, o
if (groups_amount + offset) >= limit:
break
payload = orjson.loads(str(notification.payload))
payload = json.loads(str(notification.payload))
if str(notification.entity) == NotificationEntity.SHOUT.value:
shout = payload
@@ -177,7 +177,7 @@ def get_notifications_grouped(author_id: int, after: int = 0, limit: int = 10, o
elif str(notification.entity) == "follower":
thread_id = "followers"
follower = orjson.loads(payload)
follower = json.loads(payload)
group = groups_by_thread.get(thread_id)
if group:
if str(notification.action) == "follow":
@@ -293,11 +293,11 @@ async def notifications_seen_thread(_, info, thread: str, after: int):
)
exclude = set()
for nr in removed_reaction_notifications:
reaction = orjson.loads(str(nr.payload))
reaction = json.loads(str(nr.payload))
reaction_id = reaction.get("id")
exclude.add(reaction_id)
for n in new_reaction_notifications:
reaction = orjson.loads(str(n.payload))
reaction = json.loads(str(n.payload))
reaction_id = reaction.get("id")
if (
reaction_id not in exclude

12
resolvers/rating.py
View File

@@ -15,21 +15,11 @@ from utils.logger import root_logger as logger
async def get_my_rates_comments(_, info, comments: list[int]) -> list[dict]:
"""
Получение реакций пользователя на комментарии
Args:
info: Контекст запроса
comments: Список ID комментариев
Returns:
list[dict]: Список словарей с реакциями пользователя на комментарии
Каждый словарь содержит:
- comment_id: ID комментария
- my_rate: Тип реакции (LIKE/DISLIKE)
"""
author_dict = info.context.get("author") if info.context else None
author_id = author_dict.get("id") if author_dict else None
if not author_id:
return [] # Возвращаем пустой список вместо словаря с ошибкой
return {"error": "Author not found"}
# Подзапрос для реакций текущего пользователя
rated_query = (

331
resolvers/reaction.py
View File

@@ -67,58 +67,50 @@ def add_reaction_stat_columns(q):
return q
def get_reactions_with_stat(q, limit=10, offset=0):
def get_reactions_with_stat(q, limit, offset):
"""
Execute the reaction query and retrieve reactions with statistics.
:param q: Query with reactions and statistics.
:param limit: Number of reactions to load.
:param offset: Pagination offset.
:return: List of reactions as dictionaries.
>>> get_reactions_with_stat(q, 10, 0) # doctest: +SKIP
[{'id': 1, 'body': 'Текст комментария', 'stat': {'rating': 5, 'comments_count': 3}, ...}]
:return: List of reactions.
"""
q = q.limit(limit).offset(offset)
reactions = []
with local_session() as session:
result_rows = session.execute(q)
for reaction, author, shout, comments_count, rating_stat in result_rows:
for reaction, author, shout, commented_stat, rating_stat in result_rows:
# Пропускаем реакции с отсутствующими shout или author
if not shout or not author:
logger.error(f"Пропущена реакция из-за отсутствия shout или author: {reaction.dict()}")
continue
# Преобразуем Reaction в словарь для доступа по ключу
reaction_dict = reaction.dict()
reaction_dict["created_by"] = author.dict()
reaction_dict["shout"] = shout.dict()
reaction_dict["stat"] = {"rating": rating_stat, "comments_count": comments_count}
reactions.append(reaction_dict)
reaction.created_by = author.dict()
reaction.shout = shout.dict()
reaction.stat = {"rating": rating_stat, "comments": commented_stat}
reactions.append(reaction)
return reactions
def is_featured_author(session, author_id) -> bool:
"""
Check if an author has at least one non-deleted featured article.
Check if an author has at least one featured article.
:param session: Database session.
:param author_id: Author ID.
:return: True if the author has a featured article, else False.
"""
return session.query(
session.query(Shout)
.where(Shout.authors.any(id=author_id))
.filter(Shout.featured_at.is_not(None), Shout.deleted_at.is_(None))
.exists()
session.query(Shout).where(Shout.authors.any(id=author_id)).filter(Shout.featured_at.is_not(None)).exists()
).scalar()
def check_to_feature(session, approver_id, reaction) -> bool:
"""
Make a shout featured if it receives more than 4 votes from authors.
Make a shout featured if it receives more than 4 votes.
:param session: Database session.
:param approver_id: Approver author ID.
@@ -126,78 +118,46 @@ def check_to_feature(session, approver_id, reaction) -> bool:
:return: True if shout should be featured, else False.
"""
if not reaction.reply_to and is_positive(reaction.kind):
# Проверяем, не содержит ли пост более 20% дизлайков
# Если да, то не должен быть featured независимо от количества лайков
if check_to_unfeature(session, reaction):
return False
# Собираем всех авторов, поставивших лайк
author_approvers = set()
approvers = {approver_id}
# Count the number of approvers
reacted_readers = (
session.query(Reaction.created_by)
.filter(
Reaction.shout == reaction.shout,
is_positive(Reaction.kind),
# Рейтинги (LIKE, DISLIKE) физически удаляются, поэтому фильтр deleted_at не нужен
)
.filter(Reaction.shout == reaction.shout, is_positive(Reaction.kind), Reaction.deleted_at.is_(None))
.distinct()
.all()
)
# Добавляем текущего одобряющего
approver = session.query(Author).filter(Author.id == approver_id).first()
if approver and is_featured_author(session, approver_id):
author_approvers.add(approver_id)
# Проверяем, есть ли у реагировавших авторов featured публикации
for (reader_id,) in reacted_readers:
for reader_id in reacted_readers:
if is_featured_author(session, reader_id):
author_approvers.add(reader_id)
# Публикация становится featured при наличии более 4 лайков от авторов
logger.debug(f"Публикация {reaction.shout} имеет {len(author_approvers)} лайков от авторов")
return len(author_approvers) > 4
approvers.add(reader_id)
return len(approvers) > 4
return False
def check_to_unfeature(session, reaction) -> bool:
def check_to_unfeature(session, rejecter_id, reaction) -> bool:
"""
Unfeature a shout if 20% of reactions are negative.
:param session: Database session.
:param rejecter_id: Rejecter author ID.
:param reaction: Reaction object.
:return: True if shout should be unfeatured, else False.
"""
if not reaction.reply_to:
# Проверяем соотношение дизлайков, даже если текущая реакция не дизлайк
if not reaction.reply_to and is_negative(reaction.kind):
total_reactions = (
session.query(Reaction)
.filter(
Reaction.shout == reaction.shout,
Reaction.reply_to.is_(None),
Reaction.kind.in_(RATING_REACTIONS),
# Рейтинги физически удаляются при удалении, поэтому фильтр deleted_at не нужен
Reaction.shout == reaction.shout, Reaction.kind.in_(RATING_REACTIONS), Reaction.deleted_at.is_(None)
)
.count()
)
negative_reactions = (
session.query(Reaction)
.filter(
Reaction.shout == reaction.shout,
is_negative(Reaction.kind),
Reaction.reply_to.is_(None),
# Рейтинги физически удаляются при удалении, поэтому фильтр deleted_at не нужен
)
.filter(Reaction.shout == reaction.shout, is_negative(Reaction.kind), Reaction.deleted_at.is_(None))
.count()
)
# Проверяем, составляют ли отрицательные реакции 20% или более от всех реакций
negative_ratio = negative_reactions / total_reactions if total_reactions > 0 else 0
logger.debug(
f"Публикация {reaction.shout}: {negative_reactions}/{total_reactions} отрицательных реакций ({negative_ratio:.2%})"
)
return total_reactions > 0 and negative_ratio >= 0.2
return total_reactions > 0 and (negative_reactions / total_reactions) >= 0.2
return False
@@ -236,8 +196,8 @@ async def _create_reaction(session, shout_id: int, is_author: bool, author_id: i
Create a new reaction and perform related actions such as updating counters and notification.
:param session: Database session.
:param shout_id: Shout ID.
:param is_author: Flag indicating if the user is the author of the shout.
:param info: GraphQL context info.
:param shout: Shout object.
:param author_id: Author ID.
:param reaction: Dictionary with reaction data.
:return: Dictionary with created reaction data.
@@ -257,14 +217,10 @@ async def _create_reaction(session, shout_id: int, is_author: bool, author_id: i
# Handle rating
if r.kind in RATING_REACTIONS:
# Проверяем сначала условие для unfeature (дизлайки имеют приоритет)
if check_to_unfeature(session, r):
if check_to_unfeature(session, author_id, r):
set_unfeatured(session, shout_id)
logger.info(f"Публикация {shout_id} потеряла статус featured из-за высокого процента дизлайков")
# Только если не было unfeature, проверяем условие для feature
elif check_to_feature(session, author_id, r):
await set_featured(session, shout_id)
logger.info(f"Публикация {shout_id} получила статус featured благодаря лайкам от авторов")
# Notify creation
await notify_reaction(rdict, "create")
@@ -325,10 +281,8 @@ async def create_reaction(_, info, reaction):
logger.debug(f"Creating reaction with data: {reaction_input}")
logger.debug(f"Author ID: {author_id}, Shout ID: {shout_id}")
if not author_id:
return {"error": "Author ID is required to create a reaction."}
if not shout_id:
return {"error": "Shout ID is required to create a reaction."}
if not shout_id or not author_id:
return {"error": "Shout ID and author ID are required to create a reaction."}
try:
with local_session() as session:
@@ -398,7 +352,7 @@ async def update_reaction(_, info, reaction):
result = session.execute(reaction_query).unique().first()
if result:
r, author, _shout, comments_count, rating_stat = result
r, author, shout, commented_stat, rating_stat = result
if not r or not author:
return {"error": "Invalid reaction ID or unauthorized"}
@@ -413,7 +367,7 @@ async def update_reaction(_, info, reaction):
session.commit()
r.stat = {
"comments_count": comments_count,
"commented": commented_stat,
"rating": rating_stat,
}
@@ -450,24 +404,15 @@ async def delete_reaction(_, info, reaction_id: int):
if r.created_by != author_id and "editor" not in roles:
return {"error": "Access denied"}
if r.kind == ReactionKind.COMMENT.value:
r.deleted_at = int(time.time())
update_author_stat(author.id)
session.add(r)
session.commit()
elif r.kind == ReactionKind.PROPOSE.value:
r.deleted_at = int(time.time())
session.add(r)
session.commit()
# TODO: add more reaction types here
else:
logger.debug(f"{user_id} user removing his #{reaction_id} reaction")
session.delete(r)
session.commit()
if check_to_unfeature(session, r):
set_unfeatured(session, r.shout)
logger.debug(f"{user_id} user removing his #{reaction_id} reaction")
reaction_dict = r.dict()
session.delete(r)
session.commit()
# Update author stat
if r.kind == ReactionKind.COMMENT.value:
update_author_stat(author.id)
await notify_reaction(reaction_dict, "delete")
return {"error": None, "reaction": reaction_dict}
@@ -487,16 +432,12 @@ def apply_reaction_filters(by, q):
shout_slug = by.get("shout")
if shout_slug:
q = q.filter(Shout.slug == shout_slug)
shout_id = by.get("shout_id")
if shout_id:
q = q.filter(Shout.id == shout_id)
shouts = by.get("shouts")
if shouts:
q = q.filter(Shout.slug.in_(shouts))
created_by = by.get("created_by", by.get("author_id"))
created_by = by.get("created_by")
if created_by:
q = q.filter(Author.id == created_by)
@@ -542,9 +483,7 @@ async def load_reactions_by(_, _info, by, limit=50, offset=0):
# Add statistics and apply filters
q = add_reaction_stat_columns(q)
q = apply_reaction_filters(by, q)
# Include reactions with deleted_at for building comment trees
# q = q.where(Reaction.deleted_at.is_(None))
q = q.where(Reaction.deleted_at.is_(None))
# Group and sort
q = q.group_by(Reaction.id, Author.id, Shout.id)
@@ -621,22 +560,24 @@ async def load_shout_comments(_, info, shout: int, limit=50, offset=0):
@query.field("load_comment_ratings")
async def load_comment_ratings(_, info, comment: int, limit=50, offset=0):
"""
Load ratings for a specified comment with pagination.
Load ratings for a specified comment with pagination and statistics.
:param info: GraphQL context info.
:param comment: Comment ID.
:param limit: Number of ratings to load.
:param offset: Pagination offset.
:return: List of ratings.
:return: List of reactions.
"""
q = query_reactions()
q = add_reaction_stat_columns(q)
# Filter, group, sort, limit, offset
q = q.filter(
and_(
Reaction.deleted_at.is_(None),
Reaction.reply_to == comment,
Reaction.kind.in_(RATING_REACTIONS),
Reaction.kind == ReactionKind.COMMENT.value,
)
)
q = q.group_by(Reaction.id, Author.id, Shout.id)
@@ -644,187 +585,3 @@ async def load_comment_ratings(_, info, comment: int, limit=50, offset=0):
# Retrieve and return reactions
return get_reactions_with_stat(q, limit, offset)
@query.field("load_comments_branch")
async def load_comments_branch(
_,
_info,
shout: int,
parent_id: int | None = None,
limit=10,
offset=0,
sort="newest",
children_limit=3,
children_offset=0,
):
"""
Загружает иерархические комментарии с возможностью пагинации корневых и дочерних.
:param info: GraphQL context info.
:param shout: ID статьи.
:param parent_id: ID родительского комментария (None для корневых).
:param limit: Количество комментариев для загрузки.
:param offset: Смещение для пагинации.
:param sort: Порядок сортировки ('newest', 'oldest', 'like').
:param children_limit: Максимальное количество дочерних комментариев.
:param children_offset: Смещение для дочерних комментариев.
:return: Список комментариев с дочерними.
"""
# Создаем базовый запрос
q = query_reactions()
q = add_reaction_stat_columns(q)
# Фильтруем по статье и типу (комментарии)
q = q.filter(
and_(
Reaction.deleted_at.is_(None),
Reaction.shout == shout,
Reaction.kind == ReactionKind.COMMENT.value,
)
)
# Фильтруем по родительскому ID
if parent_id is None:
# Загружаем только корневые комментарии
q = q.filter(Reaction.reply_to.is_(None))
else:
# Загружаем только прямые ответы на указанный комментарий
q = q.filter(Reaction.reply_to == parent_id)
# Сортировка и группировка
q = q.group_by(Reaction.id, Author.id, Shout.id)
# Определяем сортировку
order_by_stmt = None
if sort.lower() == "oldest":
order_by_stmt = asc(Reaction.created_at)
elif sort.lower() == "like":
order_by_stmt = desc("rating_stat")
else: # "newest" по умолчанию
order_by_stmt = desc(Reaction.created_at)
q = q.order_by(order_by_stmt)
# Выполняем запрос для получения комментариев
comments = get_reactions_with_stat(q, limit, offset)
# Если комментарии найдены, загружаем дочерние и количество ответов
if comments:
# Загружаем количество ответов для каждого комментария
await load_replies_count(comments)
# Загружаем дочерние комментарии
await load_first_replies(comments, children_limit, children_offset, sort)
return comments
async def load_replies_count(comments):
"""
Загружает количество ответов для списка комментариев и обновляет поле stat.comments_count.
:param comments: Список комментариев, для которых нужно загрузить количество ответов.
"""
if not comments:
return
comment_ids = [comment["id"] for comment in comments]
# Запрос для подсчета количества ответов
q = (
select(Reaction.reply_to.label("parent_id"), func.count().label("count"))
.where(
and_(
Reaction.reply_to.in_(comment_ids),
Reaction.deleted_at.is_(None),
Reaction.kind == ReactionKind.COMMENT.value,
)
)
.group_by(Reaction.reply_to)
)
# Выполняем запрос
with local_session() as session:
result = session.execute(q).fetchall()
# Создаем словарь {parent_id: count}
replies_count = {row[0]: row[1] for row in result}
# Добавляем значения в комментарии
for comment in comments:
if "stat" not in comment:
comment["stat"] = {}
# Обновляем счетчик комментариев в stat
comment["stat"]["comments_count"] = replies_count.get(comment["id"], 0)
async def load_first_replies(comments, limit, offset, sort="newest"):
"""
Загружает первые N ответов для каждого комментария.
:param comments: Список комментариев, для которых нужно загрузить ответы.
:param limit: Максимальное количество ответов для каждого комментария.
:param offset: Смещение для пагинации дочерних комментариев.
:param sort: Порядок сортировки ответов.
"""
if not comments or limit <= 0:
return
# Собираем ID комментариев
comment_ids = [comment["id"] for comment in comments]
# Базовый запрос для загрузки ответов
q = query_reactions()
q = add_reaction_stat_columns(q)
# Фильтрация: только ответы на указанные комментарии
q = q.filter(
and_(
Reaction.reply_to.in_(comment_ids),
Reaction.deleted_at.is_(None),
Reaction.kind == ReactionKind.COMMENT.value,
)
)
# Группировка
q = q.group_by(Reaction.id, Author.id, Shout.id)
# Определяем сортировку
order_by_stmt = None
if sort.lower() == "oldest":
order_by_stmt = asc(Reaction.created_at)
elif sort.lower() == "like":
order_by_stmt = desc("rating_stat")
else: # "newest" по умолчанию
order_by_stmt = desc(Reaction.created_at)
q = q.order_by(order_by_stmt, Reaction.reply_to)
# Выполняем запрос - указываем limit для неограниченного количества ответов
# но не более 100 на родительский комментарий
replies = get_reactions_with_stat(q, limit=100, offset=0)
# Группируем ответы по родительским ID
replies_by_parent = {}
for reply in replies:
parent_id = reply.get("reply_to")
if parent_id not in replies_by_parent:
replies_by_parent[parent_id] = []
replies_by_parent[parent_id].append(reply)
# Добавляем ответы к соответствующим комментариям с учетом смещения и лимита
for comment in comments:
comment_id = comment["id"]
if comment_id in replies_by_parent:
parent_replies = replies_by_parent[comment_id]
# Применяем смещение и лимит
comment["first_replies"] = parent_replies[offset : offset + limit]
else:
comment["first_replies"] = []
# Загружаем количество ответов для дочерних комментариев
all_replies = [reply for replies in replies_by_parent.values() for reply in replies]
if all_replies:
await load_replies_count(all_replies)

131
resolvers/reader.py
View File

@@ -1,13 +1,16 @@
import orjson
import json
import time
from graphql import GraphQLResolveInfo
from sqlalchemy import and_, nulls_last, text
from sqlalchemy import nulls_last, text
from sqlalchemy.orm import aliased
from sqlalchemy.sql.expression import asc, case, desc, func, select
from sqlalchemy.sql.expression import and_, asc, case, desc, func, select
from orm.author import Author
from orm.reaction import Reaction, ReactionKind
from orm.shout import Shout, ShoutAuthor, ShoutTopic
from orm.topic import Topic
from services.auth import login_accepted
from services.db import json_array_builder, json_builder, local_session
from services.schema import query
from services.search import search_text
@@ -61,11 +64,10 @@ def query_with_stat(info):
Добавляет подзапрос статистики
"""
q = select(Shout).filter(
and_(
Shout.published_at.is_not(None), # Проверяем published_at
Shout.deleted_at.is_(None), # Проверяем deleted_at
)
q = (
select(Shout)
.where(and_(Shout.published_at.is_not(None), Shout.deleted_at.is_(None)))
.join(Author, Author.id == Shout.created_by)
)
# Главный автор
@@ -97,6 +99,22 @@ def query_with_stat(info):
).label("main_topic")
)
if has_field(info, "topics"):
topics_subquery = (
select(
ShoutTopic.shout,
json_array_builder(
json_builder("id", Topic.id, "title", Topic.title, "slug", Topic.slug, "is_main", ShoutTopic.main)
).label("topics"),
)
.outerjoin(Topic, ShoutTopic.topic == Topic.id)
.where(ShoutTopic.shout == Shout.id)
.group_by(ShoutTopic.shout)
.subquery()
)
q = q.outerjoin(topics_subquery, topics_subquery.c.shout == Shout.id)
q = q.add_columns(topics_subquery.c.topics)
if has_field(info, "authors"):
authors_subquery = (
select(
@@ -126,23 +144,8 @@ def query_with_stat(info):
q = q.outerjoin(authors_subquery, authors_subquery.c.shout == Shout.id)
q = q.add_columns(authors_subquery.c.authors)
if has_field(info, "topics"):
topics_subquery = (
select(
ShoutTopic.shout,
json_array_builder(
json_builder("id", Topic.id, "title", Topic.title, "slug", Topic.slug, "is_main", ShoutTopic.main)
).label("topics"),
)
.outerjoin(Topic, ShoutTopic.topic == Topic.id)
.where(ShoutTopic.shout == Shout.id)
.group_by(ShoutTopic.shout)
.subquery()
)
q = q.outerjoin(topics_subquery, topics_subquery.c.shout == Shout.id)
q = q.add_columns(topics_subquery.c.topics)
if has_field(info, "stat"):
# Подзапрос для статистики реакций
stats_subquery = (
select(
Reaction.shout,
@@ -187,15 +190,17 @@ def get_shouts_with_links(info, q, limit=20, offset=0):
"""
shouts = []
try:
# logger.info(f"Starting get_shouts_with_links with limit={limit}, offset={offset}")
logger.info(f"Начало выполнения get_shouts_with_links с limit={limit}, offset={offset}")
q = q.limit(limit).offset(offset)
with local_session() as session:
logger.info("Выполнение основного запроса")
t1 = time.time()
shouts_result = session.execute(q).all()
# logger.info(f"Got {len(shouts_result) if shouts_result else 0} shouts from query")
logger.info(f"Запрос выполнен, получено {len(shouts_result)} результатов за {time.time() - t1:.3f} секунд")
if not shouts_result:
logger.warning("No shouts found in query result")
logger.warning("Нет найденных результатов")
return []
for idx, row in enumerate(shouts_result):
@@ -203,7 +208,6 @@ def get_shouts_with_links(info, q, limit=20, offset=0):
shout = None
if hasattr(row, "Shout"):
shout = row.Shout
# logger.debug(f"Processing shout#{shout.id} at index {idx}")
if shout:
shout_id = int(f"{shout.id}")
shout_dict = shout.dict()
@@ -218,71 +222,37 @@ def get_shouts_with_links(info, q, limit=20, offset=0):
"pic": a.pic,
}
if has_field(info, "stat"):
if hasattr(row, "stat"):
stat = {}
if isinstance(row.stat, str):
stat = orjson.loads(row.stat)
stat = json.loads(row.stat)
elif isinstance(row.stat, dict):
stat = row.stat
else:
logger.warning(f"Строка {idx} - неизвестный тип stat: {type(row.stat)}")
viewed = ViewedStorage.get_shout(shout_id=shout_id) or 0
shout_dict["stat"] = {**stat, "viewed": viewed}
# Обработка main_topic и topics
topics = None
if has_field(info, "topics") and hasattr(row, "topics"):
topics = orjson.loads(row.topics) if isinstance(row.topics, str) else row.topics
# logger.debug(f"Shout#{shout_id} topics: {topics}")
shout_dict["topics"] = topics
if has_field(info, "main_topic"):
main_topic = None
if hasattr(row, "main_topic"):
# logger.debug(f"Raw main_topic for shout#{shout_id}: {row.main_topic}")
main_topic = (
orjson.loads(row.main_topic) if isinstance(row.main_topic, str) else row.main_topic
)
# logger.debug(f"Parsed main_topic for shout#{shout_id}: {main_topic}")
if not main_topic and topics and len(topics) > 0:
# logger.info(f"No main_topic found for shout#{shout_id}, using first topic from list")
main_topic = {
"id": topics[0]["id"],
"title": topics[0]["title"],
"slug": topics[0]["slug"],
"is_main": True,
}
elif not main_topic:
logger.warning(f"No main_topic and no topics found for shout#{shout_id}")
main_topic = {"id": 0, "title": "no topic", "slug": "notopic", "is_main": True}
shout_dict["main_topic"] = main_topic
# logger.debug(f"Final main_topic for shout#{shout_id}: {main_topic}")
shout_dict["stat"] = {**stat, "viewed": viewed, "commented": stat.get("comments_count", 0)}
if has_field(info, "main_topic") and hasattr(row, "main_topic"):
shout_dict["main_topic"] = (
json.loads(row.main_topic) if isinstance(row.main_topic, str) else row.main_topic
)
if has_field(info, "authors") and hasattr(row, "authors"):
shout_dict["authors"] = (
orjson.loads(row.authors) if isinstance(row.authors, str) else row.authors
json.loads(row.authors) if isinstance(row.authors, str) else row.authors
)
if has_field(info, "media") and shout.media:
# Обработка поля media
media_data = shout.media
if isinstance(media_data, str):
try:
media_data = orjson.loads(media_data)
except orjson.JSONDecodeError:
media_data = []
shout_dict["media"] = [media_data] if isinstance(media_data, dict) else media_data
if has_field(info, "topics") and hasattr(row, "topics"):
shout_dict["topics"] = json.loads(row.topics) if isinstance(row.topics, str) else row.topics
shouts.append(shout_dict)
except Exception as row_error:
logger.error(f"Error processing row {idx}: {row_error}", exc_info=True)
logger.error(f"Ошибка при обработке строки {idx}: {row_error}", exc_info=True)
continue
except Exception as e:
logger.error(f"Fatal error in get_shouts_with_links: {e}", exc_info=True)
logger.error(f"Фатальная ошибка в get_shouts_with_links: {e}", exc_info=True)
raise
finally:
logger.info(f"Returning {len(shouts)} shouts from get_shouts_with_links")
return shouts
@@ -319,6 +289,7 @@ def apply_filters(q, filters):
@query.field("get_shout")
@login_accepted
async def get_shout(_, info: GraphQLResolveInfo, slug="", shout_id=0):
"""
Получение публикации по slug или id.
@@ -375,13 +346,11 @@ async def load_shouts_by(_, info: GraphQLResolveInfo, options):
:param _: Корневой объект запроса (не используется)
:param info: Информация о контексте GraphQL
:param options: Опции фильтрации и сортировки
:return: Список публикаций, удовлетворяющих критериям
:param options: Опции фильтрации и сортировки.
:return: Список публикаций, удовлетворяющих критериям.
"""
# Базовый запрос со статистикой
# Базовый запрос: используем специальный запрос с статистикой
q = query_with_stat(info)
# Применяем остальные опции фильтрации
q, limit, offset = apply_options(q, options)
# Передача сформированного запроса в метод получения публикаций с учетом сортировки и пагинации

58
resolvers/stat.py
View File

@@ -63,35 +63,47 @@ def add_author_stat_columns(q):
:param q: SQL-запрос для получения авторов.
:return: Запрос с добавленными колонками статистики.
"""
# Подзапрос для подсчета публикаций
shouts_subq = (
select(func.count(distinct(Shout.id)))
.select_from(ShoutAuthor)
.join(Shout, and_(Shout.id == ShoutAuthor.shout, Shout.deleted_at.is_(None)))
.where(ShoutAuthor.author == Author.id)
.scalar_subquery()
)
# Алиасирование таблиц для предотвращения конфликтов имен
aliased_shout_author = aliased(ShoutAuthor)
aliased_shout = aliased(Shout)
aliased_author_follower = aliased(AuthorFollower)
# Подзапрос для подсчета подписчиков
followers_subq = (
select(func.count(distinct(AuthorFollower.follower)))
.where(AuthorFollower.author == Author.id)
.scalar_subquery()
)
# Основной запрос
# Применение фильтров и добавление колонок статистики
q = (
q.select_from(Author)
.add_columns(shouts_subq.label("shouts_stat"), followers_subq.label("followers_stat"))
.group_by(Author.id)
.join(
aliased_shout_author,
aliased_shout_author.author == Author.id,
)
.join(
aliased_shout,
and_(
aliased_shout.id == aliased_shout_author.shout,
aliased_shout.deleted_at.is_(None),
),
)
.add_columns(
func.count(distinct(aliased_shout.id)).label("shouts_stat")
) # Подсчет уникальных публикаций автора
)
# Добавляем количество подписчиков автора
q = q.outerjoin(aliased_author_follower, aliased_author_follower.author == Author.id).add_columns(
func.count(distinct(aliased_author_follower.follower)).label("followers_stat")
)
# Группировка по идентификатору автора
q = q.group_by(Author.id)
return q
def get_topic_shouts_stat(topic_id: int) -> int:
"""
Получает количество опубликованных постов для темы
Получает количество публикаций для указанной темы.
:param topic_id: Идентификатор темы.
:return: Количество уникальных публикаций для темы.
"""
q = (
select(func.count(distinct(ShoutTopic.shout)))
@@ -104,7 +116,7 @@ def get_topic_shouts_stat(topic_id: int) -> int:
)
)
)
# Выполнение запроса и получение результата
with local_session() as session:
result = session.execute(q).first()
return result[0] if result else 0
@@ -186,7 +198,10 @@ def get_topic_comments_stat(topic_id: int) -> int:
def get_author_shouts_stat(author_id: int) -> int:
"""
Получает количество опубликованных постов для автора
Получает количество публикаций для указанного автора.
:param author_id: Идентификатор автора.
:return: Количество уникальных публикаций автора.
"""
aliased_shout_author = aliased(ShoutAuthor)
aliased_shout = aliased(Shout)
@@ -199,7 +214,6 @@ def get_author_shouts_stat(author_id: int) -> int:
and_(
aliased_shout_author.author == author_id,
aliased_shout.published_at.is_not(None),
aliased_shout.deleted_at.is_(None), # Добавляем проверку на удаление
)
)
)

266
resolvers/topic.py
View File

@@ -1,249 +1,44 @@
from sqlalchemy import desc, select, text
from sqlalchemy import select
from cache.cache import (
cache_topic,
cached_query,
get_cached_topic_authors,
get_cached_topic_by_slug,
get_cached_topic_followers,
invalidate_cache_by_prefix,
)
from cache.memorycache import cache_region
from orm.author import Author
from orm.topic import Topic
from orm.reaction import ReactionKind
from resolvers.stat import get_with_stat
from services.auth import login_required
from services.db import local_session
from services.redis import redis
from services.schema import mutation, query
from utils.logger import root_logger as logger
# Вспомогательная функция для получения всех тем без статистики
async def get_all_topics():
"""
Получает все темы без статистики.
Используется для случаев, когда нужен полный список тем без дополнительной информации.
Returns:
list: Список всех тем без статистики
"""
cache_key = "topics:all:basic"
# Функция для получения всех тем из БД
async def fetch_all_topics():
logger.debug("Получаем список всех тем из БД и кешируем результат")
with local_session() as session:
# Запрос на получение базовой информации о темах
topics_query = select(Topic)
topics = session.execute(topics_query).scalars().all()
# Преобразуем темы в словари
return [topic.dict() for topic in topics]
# Используем универсальную функцию для кеширования запросов
return await cached_query(cache_key, fetch_all_topics)
# Вспомогательная функция для получения тем со статистикой с пагинацией
async def get_topics_with_stats(limit=100, offset=0, community_id=None, by=None):
"""
Получает темы со статистикой с пагинацией.
Args:
limit: Максимальное количество возвращаемых тем
offset: Смещение для пагинации
community_id: Опциональный ID сообщества для фильтрации
by: Опциональный параметр сортировки
Returns:
list: Список тем с их статистикой
"""
# Формируем ключ кеша с помощью универсальной функции
cache_key = f"topics:stats:limit={limit}:offset={offset}:community_id={community_id}"
# Функция для получения тем из БД
async def fetch_topics_with_stats():
logger.debug(f"Выполняем запрос на получение тем со статистикой: limit={limit}, offset={offset}")
with local_session() as session:
# Базовый запрос для получения тем
base_query = select(Topic)
# Добавляем фильтр по сообществу, если указан
if community_id:
base_query = base_query.where(Topic.community == community_id)
# Применяем сортировку на основе параметра by
if by:
if isinstance(by, dict):
# Обработка словаря параметров сортировки
for field, direction in by.items():
column = getattr(Topic, field, None)
if column:
if direction.lower() == "desc":
base_query = base_query.order_by(desc(column))
else:
base_query = base_query.order_by(column)
elif by == "popular":
# Сортировка по популярности (количеству публикаций)
# Примечание: это требует дополнительного запроса или подзапроса
base_query = base_query.order_by(
desc(Topic.id)
) # Временно, нужно заменить на proper implementation
else:
# По умолчанию сортируем по ID в обратном порядке
base_query = base_query.order_by(desc(Topic.id))
else:
# По умолчанию сортируем по ID в обратном порядке
base_query = base_query.order_by(desc(Topic.id))
# Применяем лимит и смещение
base_query = base_query.limit(limit).offset(offset)
# Получаем темы
topics = session.execute(base_query).scalars().all()
topic_ids = [topic.id for topic in topics]
if not topic_ids:
return []
# Запрос на получение статистики по публикациям для выбранных тем
shouts_stats_query = f"""
SELECT st.topic, COUNT(DISTINCT s.id) as shouts_count
FROM shout_topic st
JOIN shout s ON st.shout = s.id AND s.deleted_at IS NULL AND s.published_at IS NOT NULL
WHERE st.topic IN ({",".join(map(str, topic_ids))})
GROUP BY st.topic
"""
shouts_stats = {row[0]: row[1] for row in session.execute(text(shouts_stats_query))}
# Запрос на получение статистики по подписчикам для выбранных тем
followers_stats_query = f"""
SELECT topic, COUNT(DISTINCT follower) as followers_count
FROM topic_followers tf
WHERE topic IN ({",".join(map(str, topic_ids))})
GROUP BY topic
"""
followers_stats = {row[0]: row[1] for row in session.execute(text(followers_stats_query))}
# Запрос на получение статистики авторов для выбранных тем
authors_stats_query = f"""
SELECT st.topic, COUNT(DISTINCT sa.author) as authors_count
FROM shout_topic st
JOIN shout s ON st.shout = s.id AND s.deleted_at IS NULL AND s.published_at IS NOT NULL
JOIN shout_author sa ON sa.shout = s.id
WHERE st.topic IN ({",".join(map(str, topic_ids))})
GROUP BY st.topic
"""
authors_stats = {row[0]: row[1] for row in session.execute(text(authors_stats_query))}
# Запрос на получение статистики комментариев для выбранных тем
comments_stats_query = f"""
SELECT st.topic, COUNT(DISTINCT r.id) as comments_count
FROM shout_topic st
JOIN shout s ON st.shout = s.id AND s.deleted_at IS NULL AND s.published_at IS NOT NULL
JOIN reaction r ON r.shout = s.id AND r.kind = '{ReactionKind.COMMENT.value}' AND r.deleted_at IS NULL
JOIN author a ON r.created_by = a.id AND a.deleted_at IS NULL
WHERE st.topic IN ({",".join(map(str, topic_ids))})
GROUP BY st.topic
"""
comments_stats = {row[0]: row[1] for row in session.execute(text(comments_stats_query))}
# Формируем результат с добавлением статистики
result = []
for topic in topics:
topic_dict = topic.dict()
topic_dict["stat"] = {
"shouts": shouts_stats.get(topic.id, 0),
"followers": followers_stats.get(topic.id, 0),
"authors": authors_stats.get(topic.id, 0),
"comments": comments_stats.get(topic.id, 0),
}
result.append(topic_dict)
# Кешируем каждую тему отдельно для использования в других функциях
await cache_topic(topic_dict)
return result
# Используем универсальную функцию для кеширования запросов
return await cached_query(cache_key, fetch_topics_with_stats)
# Функция для инвалидации кеша тем
async def invalidate_topics_cache(topic_id=None):
"""
Инвалидирует кеши тем при изменении данных.
Args:
topic_id: Опциональный ID темы для точечной инвалидации.
Если не указан, инвалидируются все кеши тем.
"""
if topic_id:
# Точечная инвалидация конкретной темы
logger.debug(f"Инвалидация кеша для темы #{topic_id}")
specific_keys = [
f"topic:id:{topic_id}",
f"topic:authors:{topic_id}",
f"topic:followers:{topic_id}",
f"topic_shouts_{topic_id}",
]
# Получаем slug темы, если есть
with local_session() as session:
topic = session.query(Topic).filter(Topic.id == topic_id).first()
if topic and topic.slug:
specific_keys.append(f"topic:slug:{topic.slug}")
# Удаляем конкретные ключи
for key in specific_keys:
try:
await redis.execute("DEL", key)
logger.debug(f"Удален ключ кеша {key}")
except Exception as e:
logger.error(f"Ошибка при удалении ключа {key}: {e}")
# Также ищем и удаляем ключи коллекций, содержащих данные об этой теме
collection_keys = await redis.execute("KEYS", "topics:stats:*")
if collection_keys:
await redis.execute("DEL", *collection_keys)
logger.debug(f"Удалено {len(collection_keys)} коллекционных ключей тем")
else:
# Общая инвалидация всех кешей тем
logger.debug("Полная инвалидация кеша тем")
await invalidate_cache_by_prefix("topics")
# Запрос на получение всех тем
@query.field("get_topics_all")
async def get_topics_all(_, _info):
"""
Получает список всех тем без статистики.
def get_topics_all(_, _info):
cache_key = "get_topics_all" # Ключ для кеша
Returns:
list: Список всех тем
"""
return await get_all_topics()
@cache_region.cache_on_arguments(cache_key)
def _get_topics_all():
topics_query = select(Topic)
return get_with_stat(topics_query) # Получение тем с учетом статистики
return _get_topics_all()
# Запрос на получение тем по сообществу
@query.field("get_topics_by_community")
async def get_topics_by_community(_, _info, community_id: int, limit=100, offset=0, by=None):
"""
Получает список тем, принадлежащих указанному сообществу с пагинацией и статистикой.
def get_topics_by_community(_, _info, community_id: int):
cache_key = f"get_topics_by_community_{community_id}" # Ключ для кеша
Args:
community_id: ID сообщества
limit: Максимальное количество возвращаемых тем
offset: Смещение для пагинации
by: Опциональные параметры сортировки
@cache_region.cache_on_arguments(cache_key)
def _get_topics_by_community():
topics_by_community_query = select(Topic).where(Topic.community == community_id)
return get_with_stat(topics_by_community_query)
Returns:
list: Список тем с их статистикой
"""
return await get_topics_with_stats(limit, offset, community_id, by)
return _get_topics_by_community()
# Запрос на получение тем по автору
@@ -271,43 +66,31 @@ async def get_topic(_, _info, slug: str):
# Мутация для создания новой темы
@mutation.field("create_topic")
@login_required
async def create_topic(_, _info, topic_input):
async def create_topic(_, _info, inp):
with local_session() as session:
# TODO: проверить права пользователя на создание темы для конкретного сообщества
# и разрешение на создание
new_topic = Topic(**topic_input)
new_topic = Topic(**inp)
session.add(new_topic)
session.commit()
# Инвалидируем кеш всех тем
await invalidate_topics_cache()
return {"topic": new_topic}
# Мутация для обновления темы
@mutation.field("update_topic")
@login_required
async def update_topic(_, _info, topic_input):
slug = topic_input["slug"]
async def update_topic(_, _info, inp):
slug = inp["slug"]
with local_session() as session:
topic = session.query(Topic).filter(Topic.slug == slug).first()
if not topic:
return {"error": "topic not found"}
else:
old_slug = topic.slug
Topic.update(topic, topic_input)
Topic.update(topic, inp)
session.add(topic)
session.commit()
# Инвалидируем кеш только для этой конкретной темы
await invalidate_topics_cache(topic.id)
# Если slug изменился, удаляем старый ключ
if old_slug != topic.slug:
await redis.execute("DEL", f"topic:slug:{old_slug}")
logger.debug(f"Удален ключ кеша для старого slug: {old_slug}")
return {"topic": topic}
@@ -328,11 +111,6 @@ async def delete_topic(_, info, slug: str):
session.delete(t)
session.commit()
# Инвалидируем кеш всех тем и конкретной темы
await invalidate_topics_cache()
await redis.execute("DEL", f"topic:slug:{slug}")
await redis.execute("DEL", f"topic:id:{t.id}")
return {}
return {"error": "access denied"}

53
schema/input.graphql
View File

@@ -1,46 +1,15 @@
input MediaItemInput {
url: String
title: String
body: String
source: String
pic: String
date: String
genre: String
artist: String
lyrics: String
}
input AuthorInput {
id: Int!
input ShoutInput {
slug: String
}
input TopicInput {
id: Int
slug: String!
title: String
body: String
pic: String
}
input DraftInput {
id: Int
# no created_at, updated_at, deleted_at, updated_by, deleted_by
layout: String
shout_id: Int # Changed from shout: Shout
author_ids: [Int!] # Changed from authors: [Author]
topic_ids: [Int!] # Changed from topics: [Topic]
main_topic_id: Int # Changed from main_topic: Topic
media: [MediaItemInput] # Changed to use MediaItemInput
lead: String
description: String
layout: String
media: String
topics: [TopicInput]
community: Int
subtitle: String
lang: String
seo: String
body: String
title: String
slug: String
cover: String
cover_caption: String
}
input ProfileInput {
@@ -52,6 +21,14 @@ input ProfileInput {
about: String
}
input TopicInput {
id: Int
slug: String!
title: String
body: String
pic: String
}
input ReactionInput {
id: Int
kind: ReactionKind!
@@ -92,14 +69,12 @@ input LoadShoutsOptions {
input ReactionBy {
shout: String
shout_id: Int
shouts: [String]
search: String
kinds: [ReactionKind]
reply_to: Int # filter
topic: String
created_by: Int
author_id: Int
author: String
after: Int
sort: ReactionSort # sort

21
schema/mutation.graphql
View File

@@ -3,23 +3,18 @@ type Mutation {
rate_author(rated_slug: String!, value: Int!): CommonResult!
update_author(profile: ProfileInput!): CommonResult!
# draft
create_draft(draft_input: DraftInput!): CommonResult!
update_draft(draft_id: Int!, draft_input: DraftInput!): CommonResult!
delete_draft(draft_id: Int!): CommonResult!
# publication
publish_shout(shout_id: Int!): CommonResult!
publish_draft(draft_id: Int!): CommonResult!
unpublish_draft(draft_id: Int!): CommonResult!
unpublish_shout(shout_id: Int!): CommonResult!
# editor
create_shout(inp: ShoutInput!): CommonResult!
update_shout(shout_id: Int!, shout_input: ShoutInput, publish: Boolean): CommonResult!
delete_shout(shout_id: Int!): CommonResult!
# follower
follow(what: FollowingEntity!, slug: String!): AuthorFollowsResult!
unfollow(what: FollowingEntity!, slug: String!): AuthorFollowsResult!
# topic
create_topic(topic_input: TopicInput!): CommonResult!
update_topic(topic_input: TopicInput!): CommonResult!
create_topic(input: TopicInput!): CommonResult!
update_topic(input: TopicInput!): CommonResult!
delete_topic(slug: String!): CommonResult!
# reaction
@@ -45,7 +40,7 @@ type Mutation {
# community
join_community(slug: String!): CommonResult!
leave_community(slug: String!): CommonResult!
create_community(community_input: CommunityInput!): CommonResult!
update_community(community_input: CommunityInput!): CommonResult!
create_community(input: CommunityInput!): CommonResult!
update_community(input: CommunityInput!): CommonResult!
delete_community(slug: String!): CommonResult!
}

6
schema/query.graphql
View File

@@ -26,9 +26,6 @@ type Query {
load_shout_ratings(shout: Int!, limit: Int, offset: Int): [Reaction]
load_comment_ratings(comment: Int!, limit: Int, offset: Int): [Reaction]
# branched comments pagination
load_comments_branch(shout: Int!, parent_id: Int, limit: Int, offset: Int, sort: ReactionSort, children_limit: Int, children_offset: Int): [Reaction]
# reader
get_shout(slug: String, shout_id: Int): Shout
load_shouts_by(options: LoadShoutsOptions): [Shout]
@@ -54,13 +51,12 @@ type Query {
# editor
get_my_shout(shout_id: Int!): CommonResult!
get_shouts_drafts: CommonResult!
load_drafts: CommonResult!
# topic
get_topic(slug: String!): Topic
get_topics_all: [Topic]
get_topics_by_author(slug: String, user: String, author_id: Int): [Topic]
get_topics_by_community(community_id: Int!, limit: Int, offset: Int): [Topic]
get_topics_by_community(slug: String, community_id: Int): [Topic]
# notifier
load_notifications(after: Int!, limit: Int, offset: Int): NotificationsResult!

44
schema/type.graphql
View File

@@ -55,7 +55,6 @@ type Reaction {
stat: Stat
oid: String
# old_thread: String
first_replies: [Reaction]
}
type MediaItem {
@@ -80,6 +79,7 @@ type Shout {
layout: String!
lead: String
description: String
subtitle: String
lang: String
cover: String
@@ -99,50 +99,16 @@ type Shout {
featured_at: Int
deleted_at: Int
seo: String # generated if not set
version_of: Shout # TODO: use version_of somewhere
draft: Draft
media: [MediaItem]
# media: [MediaItem]
stat: Stat
score: Float
}
type PublicationInfo {
id: Int!
slug: String!
published_at: Int
}
type Draft {
id: Int!
created_at: Int!
created_by: Author!
community: Community!
layout: String
slug: String
title: String
subtitle: String
lead: String
body: String
media: [MediaItem]
cover: String
cover_caption: String
lang: String
seo: String
# auto
updated_at: Int
deleted_at: Int
updated_by: Author
deleted_by: Author
authors: [Author]!
topics: [Topic]!
publication: PublicationInfo
}
type Stat {
rating: Int
comments_count: Int
commented: Int
viewed: Int
last_commented_at: Int
}
@@ -197,8 +163,6 @@ type Topic {
type CommonResult {
error: String
drafts: [Draft]
draft: Draft
slugs: [String]
shout: Shout
shouts: [Shout]

34
server.py Normal file
View File

@@ -0,0 +1,34 @@
import sys
from pathlib import Path
from granian.constants import Interfaces
from granian.log import LogLevels
from granian.server import Granian
from settings import PORT
from utils.logger import root_logger as logger
if __name__ == "__main__":
logger.info("started")
try:
granian_instance = Granian(
"main:app",
address="0.0.0.0",
port=PORT,
interface=Interfaces.ASGI,
threads=4,
websockets=False,
log_level=LogLevels.debug,
backlog=2048,
)
if "dev" in sys.argv:
logger.info("dev mode, building ssl context")
granian_instance.build_ssl_context(cert=Path("localhost.pem"), key=Path("localhost-key.pem"), password=None)
granian_instance.serve()
except Exception as error:
logger.error(f"Granian error: {error}", exc_info=True)
raise
finally:
logger.info("stopped")

1
services/__init__.py
View File

@@ -1 +0,0 @@
# This file makes services a Python package

28
services/auth.py
View File

@@ -7,8 +7,10 @@ from settings import ADMIN_SECRET, AUTH_URL
from utils.logger import root_logger as logger
# Список разрешенных заголовков
ALLOWED_HEADERS = ["Authorization", "Content-Type"]
ALLOWED_HEADERS = [
'Authorization',
'Content-Type'
]
async def check_auth(req):
"""
@@ -25,26 +27,32 @@ async def check_auth(req):
- user_roles: list[str] - Список ролей пользователя.
"""
token = req.headers.get("Authorization")
host = req.headers.get("host", "")
host = req.headers.get('host', '')
logger.debug(f"check_auth: host={host}")
auth_url = AUTH_URL
if ".dscrs.site" in host or "localhost" in host:
if '.dscrs.site' in host or 'localhost' in host:
auth_url = "https://auth.dscrs.site/graphql"
user_id = ""
user_roles = []
if token:
# Проверяем и очищаем токен от префикса Bearer если он есть
if token.startswith("Bearer "):
token = token.split("Bearer ")[-1].strip()
if token.startswith('Bearer '):
token = token.split('Bearer ')[-1].strip()
# Logging the authentication token
logger.debug(f"TOKEN: {token}")
query_name = "validate_jwt_token"
operation = "ValidateToken"
variables = {"params": {"token_type": "access_token", "token": token}}
# Только необходимые заголовки для GraphQL запроса
headers = {"Content-Type": "application/json"}
# Добавляем CORS заголовки
headers = {
'Content-Type': 'application/json',
'Access-Control-Allow-Origin': '*',
'Access-Control-Allow-Methods': 'GET, POST, OPTIONS',
'Access-Control-Allow-Headers': ', '.join(ALLOWED_HEADERS),
'Access-Control-Allow-Credentials': 'true'
}
gql = {
"query": f"query {operation}($params: ValidateJWTTokenInput!)"
@@ -106,7 +114,7 @@ def login_required(f):
"""
Декоратор для проверки авторизации пользователя.
Этот декоратор проверяет, авторизован ли пользователь, <EFBFBD><EFBFBD> добавляет
Этот декоратор проверяет, авторизован ли пользователь, и добавляет
информацию о пользователе в контекст функции.
Параметры:

32
services/common_result.py
View File

@@ -1,5 +1,5 @@
from dataclasses import dataclass
from typing import List, Optional
from typing import Any, Dict, List, Optional
from orm.author import Author
from orm.community import Community
@@ -22,3 +22,33 @@ class CommonResult:
topics: Optional[List[Topic]] = None
community: Optional[Community] = None
communities: Optional[List[Community]] = None
@classmethod
def ok(cls, data: Dict[str, Any]) -> "CommonResult":
"""
Создает успешный результат.
Args:
data: Словарь с данными для включения в результат.
Returns:
CommonResult: Экземпляр с предоставленными данными.
"""
result = cls()
for key, value in data.items():
if hasattr(result, key):
setattr(result, key, value)
return result
@classmethod
def error(cls, message: str):
"""
Create an error result.
Args:
message: The error message.
Returns:
CommonResult: An instance with the error message.
"""
return cls(error=message)

99
services/db.py
View File

@@ -1,25 +1,14 @@
import json
import math
import time
import traceback
import warnings
from typing import Any, Callable, Dict, TypeVar
import orjson
import sqlalchemy
from sqlalchemy import (
JSON,
Column,
Engine,
Index,
Integer,
create_engine,
event,
exc,
func,
inspect,
text,
)
from sqlalchemy.orm import Session, configure_mappers, declarative_base
from sqlalchemy import JSON, Column, Engine, Integer, create_engine, event, exc, func, inspect
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session, configure_mappers
from sqlalchemy.sql.schema import Table
from settings import DB_URL
@@ -58,82 +47,6 @@ def create_table_if_not_exists(engine, table):
logger.info(f"Table '{table.__tablename__}' ok.")
def sync_indexes():
"""
Синхронизирует индексы в БД с индексами, определенными в моделях SQLAlchemy.
Создает недостающие индексы, если они определены в моделях, но отсутствуют в БД.
Использует pg_catalog для PostgreSQL для получения списка существующих индексов.
"""
if not DB_URL.startswith("postgres"):
logger.warning("Функция sync_indexes поддерживается только для PostgreSQL.")
return
logger.info("Начинаем синхронизацию индексов в базе данных...")
# Получаем все существующие индексы в БД
with local_session() as session:
existing_indexes_query = text("""
SELECT
t.relname AS table_name,
i.relname AS index_name
FROM
pg_catalog.pg_class i
JOIN
pg_catalog.pg_index ix ON ix.indexrelid = i.oid
JOIN
pg_catalog.pg_class t ON t.oid = ix.indrelid
JOIN
pg_catalog.pg_namespace n ON n.oid = i.relnamespace
WHERE
i.relkind = 'i'
AND n.nspname = 'public'
AND t.relkind = 'r'
ORDER BY
t.relname, i.relname;
""")
existing_indexes = {row[1].lower() for row in session.execute(existing_indexes_query)}
logger.debug(f"Найдено {len(existing_indexes)} существующих индексов в БД")
# Проверяем каждую модель и её индексы
for _model_name, model_class in REGISTRY.items():
if hasattr(model_class, "__table__") and hasattr(model_class, "__table_args__"):
table_args = model_class.__table_args__
# Если table_args - это кортеж, ищем в нём объекты Index
if isinstance(table_args, tuple):
for arg in table_args:
if isinstance(arg, Index):
index_name = arg.name.lower()
# Проверяем, существует ли индекс в БД
if index_name not in existing_indexes:
logger.info(
f"Создаем отсутствующий индекс {index_name} для таблицы {model_class.__tablename__}"
)
# Создаем индекс если он отсутствует
try:
arg.create(engine)
logger.info(f"Индекс {index_name} успешно создан")
except Exception as e:
logger.error(f"Ошибка при создании индекса {index_name}: {e}")
else:
logger.debug(f"Индекс {index_name} уже существует")
# Анализируем таблицы для оптимизации запросов
for model_name, model_class in REGISTRY.items():
if hasattr(model_class, "__tablename__"):
try:
session.execute(text(f"ANALYZE {model_class.__tablename__}"))
logger.debug(f"Таблица {model_class.__tablename__} проанализирована")
except Exception as e:
logger.error(f"Ошибка при анализе таблицы {model_class.__tablename__}: {e}")
logger.info("Синхронизация индексов завершена.")
# noinspection PyUnusedLocal
def local_session(src=""):
return Session(bind=engine, expire_on_commit=False)
@@ -162,8 +75,8 @@ class Base(declarative_base()):
# Check if the value is JSON and decode it if necessary
if isinstance(value, (str, bytes)) and isinstance(self.__table__.columns[column_name].type, JSON):
try:
data[column_name] = orjson.loads(value)
except (TypeError, orjson.JSONDecodeError) as e:
data[column_name] = json.loads(value)
except (TypeError, json.JSONDecodeError) as e:
logger.error(f"Error decoding JSON for column '{column_name}': {e}")
data[column_name] = value
else:

76
services/notify.py
View File

@@ -1,4 +1,4 @@
import orjson
import json
from orm.notification import Notification
from services.db import local_session
@@ -18,7 +18,7 @@ async def notify_reaction(reaction, action: str = "create"):
data = {"payload": reaction, "action": action}
try:
save_notification(action, channel_name, data.get("payload"))
await redis.publish(channel_name, orjson.dumps(data))
await redis.publish(channel_name, json.dumps(data))
except Exception as e:
logger.error(f"Failed to publish to channel {channel_name}: {e}")
@@ -28,7 +28,7 @@ async def notify_shout(shout, action: str = "update"):
data = {"payload": shout, "action": action}
try:
save_notification(action, channel_name, data.get("payload"))
await redis.publish(channel_name, orjson.dumps(data))
await redis.publish(channel_name, json.dumps(data))
except Exception as e:
logger.error(f"Failed to publish to channel {channel_name}: {e}")
@@ -43,7 +43,7 @@ async def notify_follower(follower: dict, author_id: int, action: str = "follow"
save_notification(action, channel_name, data.get("payload"))
# Convert data to JSON string
json_data = orjson.dumps(data)
json_data = json.dumps(data)
# Ensure the data is not empty before publishing
if json_data:
@@ -53,71 +53,3 @@ async def notify_follower(follower: dict, author_id: int, action: str = "follow"
except Exception as e:
# Log the error and re-raise it
logger.error(f"Failed to publish to channel {channel_name}: {e}")
async def notify_draft(draft_data, action: str = "publish"):
"""
Отправляет уведомление о публикации или обновлении черновика.
Функция гарантирует, что данные черновика сериализуются корректно, включая
связанные атрибуты (topics, authors).
Args:
draft_data (dict): Словарь с данными черновика. Должен содержать минимум id и title
action (str, optional): Действие ("publish", "update"). По умолчанию "publish"
Returns:
None
Examples:
>>> draft = {"id": 1, "title": "Тестовый черновик", "slug": "test-draft"}
>>> await notify_draft(draft, "publish")
"""
channel_name = "draft"
try:
# Убеждаемся, что все необходимые данные присутствуют
# и объект не требует доступа к отсоединенным атрибутам
if isinstance(draft_data, dict):
draft_payload = draft_data
else:
# Если это ORM объект, преобразуем его в словарь с нужными атрибутами
draft_payload = {
"id": getattr(draft_data, "id", None),
"slug": getattr(draft_data, "slug", None),
"title": getattr(draft_data, "title", None),
"subtitle": getattr(draft_data, "subtitle", None),
"media": getattr(draft_data, "media", None),
"created_at": getattr(draft_data, "created_at", None),
"updated_at": getattr(draft_data, "updated_at", None),
}
# Если переданы связанные атрибуты, добавим их
if hasattr(draft_data, "topics") and draft_data.topics is not None:
draft_payload["topics"] = [
{"id": t.id, "name": t.name, "slug": t.slug}
for t in draft_data.topics
]
if hasattr(draft_data, "authors") and draft_data.authors is not None:
draft_payload["authors"] = [
{
"id": a.id,
"name": a.name,
"slug": a.slug,
"pic": getattr(a, "pic", None),
}
for a in draft_data.authors
]
data = {"payload": draft_payload, "action": action}
# Сохраняем уведомление
save_notification(action, channel_name, data.get("payload"))
# Публикуем в Redis
json_data = orjson.dumps(data)
if json_data:
await redis.publish(channel_name, json_data)
except Exception as e:
logger.error(f"Failed to publish to channel {channel_name}: {e}")

170
services/pretopic.py
View File

@@ -1,170 +0,0 @@
import concurrent.futures
from typing import Dict, List, Tuple
from txtai.embeddings import Embeddings
from services.logger import root_logger as logger
class TopicClassifier:
def __init__(self, shouts_by_topic: Dict[str, str], publications: List[Dict[str, str]]):
"""
Инициализация классификатора тем и поиска публикаций.
Args:
shouts_by_topic: Словарь {тема: текст_всех_публикаций}
publications: Список публикаций с полями 'id', 'title', 'text'
"""
self.shouts_by_topic = shouts_by_topic
self.topics = list(shouts_by_topic.keys())
self.publications = publications
self.topic_embeddings = None # Для классификации тем
self.search_embeddings = None # Для поиска публикаций
self._initialization_future = None
self._executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
def initialize(self) -> None:
"""
Асинхронная инициализация векторных представлений.
"""
if self._initialization_future is None:
self._initialization_future = self._executor.submit(self._prepare_embeddings)
logger.info("Векторизация текстов начата в фоновом режиме...")
def _prepare_embeddings(self) -> None:
"""
Подготавливает векторные представления для тем и поиска.
"""
logger.info("Начинается подготовка векторных представлений...")
# Модель для русского языка
# TODO: model local caching
model_path = "sentence-transformers/paraphrase-multilingual-mpnet-base-v2"
# Инициализируем embeddings для классификации тем
self.topic_embeddings = Embeddings(path=model_path)
topic_documents = [(topic, text) for topic, text in self.shouts_by_topic.items()]
self.topic_embeddings.index(topic_documents)
# Инициализируем embeddings для поиска публикаций
self.search_embeddings = Embeddings(path=model_path)
search_documents = [(str(pub["id"]), f"{pub['title']} {pub['text']}") for pub in self.publications]
self.search_embeddings.index(search_documents)
logger.info("Подготовка векторных представлений завершена.")
def predict_topic(self, text: str) -> Tuple[float, str]:
"""
Предсказывает тему для заданного текста из известного набора тем.
Args:
text: Текст для классификации
Returns:
Tuple[float, str]: (уверенность, тема)
"""
if not self.is_ready():
logger.error("Векторные представления не готовы. Вызовите initialize() и дождитесь завершения.")
return 0.0, "unknown"
try:
# Ищем наиболее похожую тему
results = self.topic_embeddings.search(text, 1)
if not results:
return 0.0, "unknown"
score, topic = results[0]
return float(score), topic
except Exception as e:
logger.error(f"Ошибка при определении темы: {str(e)}")
return 0.0, "unknown"
def search_similar(self, query: str, limit: int = 5) -> List[Dict[str, any]]:
"""
Ищет публикации похожие на поисковый запрос.
Args:
query: Поисковый запрос
limit: Максимальное количество результатов
Returns:
List[Dict]: Список найденных публикаций с оценкой релевантности
"""
if not self.is_ready():
logger.error("Векторные представления не готовы. Вызовите initialize() и дождитесь завершения.")
return []
try:
# Ищем похожие публикации
results = self.search_embeddings.search(query, limit)
# Формируем результаты
found_publications = []
for score, pub_id in results:
# Находим публикацию по id
publication = next((pub for pub in self.publications if str(pub["id"]) == pub_id), None)
if publication:
found_publications.append({**publication, "relevance": float(score)})
return found_publications
except Exception as e:
logger.error(f"Ошибка при поиске публикаций: {str(e)}")
return []
def is_ready(self) -> bool:
"""
Проверяет, готовы ли векторные представления.
"""
return self.topic_embeddings is not None and self.search_embeddings is not None
def wait_until_ready(self) -> None:
"""
Ожидает завершения подготовки векторных представлений.
"""
if self._initialization_future:
self._initialization_future.result()
def __del__(self):
"""
Очистка ресурсов при удалении объекта.
"""
if self._executor:
self._executor.shutdown(wait=False)
# Пример использования:
"""
shouts_by_topic = {
"Спорт": "... большой текст со всеми спортивными публикациями ...",
"Технологии": "... большой текст со всеми технологическими публикациями ...",
"Политика": "... большой текст со всеми политическими публикациями ..."
}
publications = [
{
'id': 1,
'title': 'Новый процессор AMD',
'text': 'Компания AMD представила новый процессор...'
},
{
'id': 2,
'title': 'Футбольный матч',
'text': 'Вчера состоялся решающий матч...'
}
]
# Создание классификатора
classifier = TopicClassifier(shouts_by_topic, publications)
classifier.initialize()
classifier.wait_until_ready()
# Определение темы текста
text = "Новый процессор показал высокую производительность"
score, topic = classifier.predict_topic(text)
print(f"Тема: {topic} (уверенность: {score:.4f})")
# Поиск похожих публикаций
query = "процессор AMD производительность"
similar_publications = classifier.search_similar(query, limit=3)
for pub in similar_publications:
print(f"\nНайдена публикация (релевантность: {pub['relevance']:.4f}):")
print(f"Заголовок: {pub['title']}")
print(f"Текст: {pub['text'][:100]}...")
"""

19
services/pyrightconfig.json
View File

@@ -1,19 +0,0 @@
{
"include": ["."],
"exclude": ["**/node_modules", "**/__pycache__", "**/.*"],
"defineConstant": {
"DEBUG": true
},
"venvPath": ".",
"venv": ".venv",
"pythonVersion": "3.11",
"typeCheckingMode": "strict",
"reportMissingImports": true,
"reportMissingTypeStubs": false,
"reportUnknownMemberType": false,
"reportUnknownParameterType": false,
"reportUnknownVariableType": false,
"reportUnknownArgumentType": false,
"reportPrivateUsage": false,
"reportUntypedFunctionDecorator": false
}

60
services/schema.py
View File

@@ -1,31 +1,22 @@
from asyncio.log import logger
import httpx
from ariadne import MutationType, ObjectType, QueryType
from ariadne import MutationType, QueryType
from services.db import create_table_if_not_exists, local_session
from settings import AUTH_URL
query = QueryType()
mutation = MutationType()
type_draft = ObjectType("Draft")
resolvers = [query, mutation, type_draft]
resolvers = [query, mutation]
async def request_graphql_data(gql, url=AUTH_URL, headers=None):
"""
Выполняет GraphQL запрос к указанному URL
:param gql: GraphQL запрос
:param url: URL для запроса, по умолчанию AUTH_URL
:param headers: Заголовки запроса
:return: Результат запроса или None в случае ошибки
"""
if not url:
return None
if headers is None:
headers = {"Content-Type": "application/json"}
try:
# logger.debug(f"{url}:\n{headers}\n{gql}")
async with httpx.AsyncClient() as client:
response = await client.post(url, json=gql, headers=headers)
if response.status_code == 200:
@@ -38,51 +29,8 @@ async def request_graphql_data(gql, url=AUTH_URL, headers=None):
else:
logger.error(f"{url}: {response.status_code} {response.text}")
except Exception as _e:
# Handling and logging exceptions during authentication check
import traceback
logger.error(f"request_graphql_data error: {traceback.format_exc()}")
return None
def create_all_tables():
"""Create all database tables in the correct order."""
from orm import author, community, draft, notification, reaction, shout, topic
# Порядок важен - сначала таблицы без внешних ключей, затем зависимые таблицы
models_in_order = [
# user.User, # Базовая таблица auth
author.Author, # Базовая таблица
community.Community, # Базовая таблица
topic.Topic, # Базовая таблица
# Связи для базовых таблиц
author.AuthorFollower, # Зависит от Author
community.CommunityFollower, # Зависит от Community
topic.TopicFollower, # Зависит от Topic
# Черновики (теперь без зависимости от Shout)
draft.Draft, # Зависит только от Author
draft.DraftAuthor, # Зависит от Draft и Author
draft.DraftTopic, # Зависит от Draft и Topic
# Основные таблицы контента
shout.Shout, # Зависит от Author и Draft
shout.ShoutAuthor, # Зависит от Shout и Author
shout.ShoutTopic, # Зависит от Shout и Topic
# Реакции
reaction.Reaction, # Зависит от Author и Shout
shout.ShoutReactionsFollower, # Зависит от Shout и Reaction
# Дополнительные таблицы
author.AuthorRating, # Зависит от Author
notification.Notification, # Зависит от Author
notification.NotificationSeen, # Зависит от Notification
# collection.Collection,
# collection.ShoutCollection,
# invite.Invite
]
with local_session() as session:
for model in models_in_order:
try:
create_table_if_not_exists(session.get_bind(), model)
# logger.info(f"Created or verified table: {model.__tablename__}")
except Exception as e:
logger.error(f"Error creating table {model.__tablename__}: {e}")
raise

3
services/search.py
View File

@@ -3,7 +3,6 @@ import json
import logging
import os
import orjson
from opensearchpy import OpenSearch
from services.redis import redis
@@ -143,7 +142,7 @@ class SearchService:
# Проверка и обновление структуры индекса, если необходимо
result = self.client.indices.get_mapping(index=self.index_name)
if isinstance(result, str):
result = orjson.loads(result)
result = json.loads(result)
if isinstance(result, dict):
mapping = result.get(self.index_name, {}).get("mappings")
logger.info(f"Найдена структура индексации: {mapping['properties'].keys()}")

145
services/viewed.py
View File

@@ -1,56 +1,46 @@
import asyncio
import json
import os
import time
from datetime import datetime, timedelta, timezone
from typing import Dict, Optional
from typing import Dict
# ga
from google.analytics.data_v1beta import BetaAnalyticsDataClient
from google.analytics.data_v1beta.types import (
DateRange,
Dimension,
Metric,
RunReportRequest,
)
from google.analytics.data_v1beta.types import DateRange, Dimension, Metric, RunReportRequest
from google.analytics.data_v1beta.types import Filter as GAFilter
from orm.author import Author
from orm.shout import Shout, ShoutAuthor, ShoutTopic
from orm.topic import Topic
from services.db import local_session
from services.redis import redis
from utils.logger import root_logger as logger
GOOGLE_KEYFILE_PATH = os.environ.get("GOOGLE_KEYFILE_PATH", "/dump/google-service.json")
GOOGLE_PROPERTY_ID = os.environ.get("GOOGLE_PROPERTY_ID", "")
VIEWS_FILEPATH = "/dump/views.json"
class ViewedStorage:
"""
Класс для хранения и доступа к данным о просмотрах.
Использует Redis в качестве основного хранилища и Google Analytics для сбора новых данных.
"""
lock = asyncio.Lock()
precounted_by_slug = {}
views_by_shout = {}
shouts_by_topic = {}
shouts_by_author = {}
views = None
period = 60 * 60 # каждый час
analytics_client: Optional[BetaAnalyticsDataClient] = None
analytics_client: BetaAnalyticsDataClient | None = None
auth_result = None
running = False
redis_views_key = None
last_update_timestamp = 0
start_date = datetime.now().strftime("%Y-%m-%d")
@staticmethod
async def init():
"""Подключение к клиенту Google Analytics и загрузка данных о просмотрах из Redis"""
"""Подключение к клиенту Google Analytics с использованием аутентификации"""
self = ViewedStorage
async with self.lock:
# Загрузка предварительно подсчитанных просмотров из Redis
await self.load_views_from_redis()
# Загрузка предварительно подсчитанных просмотров из файла JSON
self.load_precounted_views()
os.environ.setdefault("GOOGLE_APPLICATION_CREDENTIALS", GOOGLE_KEYFILE_PATH)
if GOOGLE_KEYFILE_PATH and os.path.isfile(GOOGLE_KEYFILE_PATH):
@@ -66,54 +56,40 @@ class ViewedStorage:
self.running = False
@staticmethod
async def load_views_from_redis():
"""Загрузка предварительно подсчитанных просмотров из Redis"""
def load_precounted_views():
"""Загрузка предварительно подсчитанных просмотров из файла JSON"""
self = ViewedStorage
viewfile_path = VIEWS_FILEPATH
if not os.path.exists(viewfile_path):
viewfile_path = os.path.join(os.path.curdir, "views.json")
if not os.path.exists(viewfile_path):
logger.warning(" * views.json not found")
return
# Подключаемся к Redis если соединение не установлено
if not redis._client:
await redis.connect()
# Получаем список всех ключей migrated_views_* и находим самый последний
keys = await redis.execute("KEYS", "migrated_views_*")
if not keys:
logger.warning(" * No migrated_views keys found in Redis")
return
# Фильтруем только ключи timestamp формата (исключаем migrated_views_slugs)
timestamp_keys = [k for k in keys if k != "migrated_views_slugs"]
if not timestamp_keys:
logger.warning(" * No migrated_views timestamp keys found in Redis")
return
# Сортируем по времени создания (в названии ключа) и берем последний
timestamp_keys.sort()
latest_key = timestamp_keys[-1]
self.redis_views_key = latest_key
# Получаем метку времени создания для установки start_date
timestamp = await redis.execute("HGET", latest_key, "_timestamp")
if timestamp:
self.last_update_timestamp = int(timestamp)
timestamp_dt = datetime.fromtimestamp(int(timestamp))
self.start_date = timestamp_dt.strftime("%Y-%m-%d")
# Если данные сегодняшние, считаем их актуальными
logger.info(f" * loading views from {viewfile_path}")
try:
start_date_int = os.path.getmtime(viewfile_path)
start_date_str = datetime.fromtimestamp(start_date_int).strftime("%Y-%m-%d")
self.start_date = start_date_str
now_date = datetime.now().strftime("%Y-%m-%d")
if now_date == self.start_date:
logger.info(" * Views data is up to date!")
else:
logger.warning(f" * Views data is from {self.start_date}, may need update")
# Выводим информацию о количестве загруженных записей
total_entries = await redis.execute("HGET", latest_key, "_total")
if total_entries:
logger.info(f" * {total_entries} shouts with views loaded from Redis key: {latest_key}")
if now_date == self.start_date:
logger.info(" * views data is up to date!")
else:
logger.warn(f" * {viewfile_path} is too old: {self.start_date}")
with open(viewfile_path, "r") as file:
precounted_views = json.load(file)
self.precounted_by_slug.update(precounted_views)
logger.info(f" * {len(precounted_views)} shouts with views was loaded.")
except Exception as e:
logger.error(f"precounted views loading error: {e}")
# noinspection PyTypeChecker
@staticmethod
async def update_pages():
"""Запрос всех страниц от Google Analytics, отсортированных по количеству просмотров"""
"""Запрос всех страниц от Google Analytics, отсортрованных по количеству просмотров"""
self = ViewedStorage
logger.info(" ⎧ views update from Google Analytics ---")
if self.running:
@@ -158,40 +134,15 @@ class ViewedStorage:
self.running = False
@staticmethod
async def get_shout(shout_slug="", shout_id=0) -> int:
"""
Получение метрики просмотров shout по slug или id.
Args:
shout_slug: Slug публикации
shout_id: ID публикации
Returns:
int: Количество просмотров
"""
def get_shout(shout_slug="", shout_id=0) -> int:
"""Получение метрики просмотров shout по slug или id."""
self = ViewedStorage
# Получаем данные из Redis для новой схемы хранения
if not redis._client:
await redis.connect()
fresh_views = self.views_by_shout.get(shout_slug, 0)
# Если есть id, пытаемся получить данные из Redis по ключу migrated_views_<timestamp>
if shout_id and self.redis_views_key:
precounted_views = await redis.execute("HGET", self.redis_views_key, str(shout_id))
if precounted_views:
return fresh_views + int(precounted_views)
# Если нет id или данных, пытаемся получить по slug из отдельного хеша
precounted_views = await redis.execute("HGET", "migrated_views_slugs", shout_slug)
if precounted_views:
return fresh_views + int(precounted_views)
return fresh_views
precounted_views = self.precounted_by_slug.get(shout_slug, 0)
return fresh_views + precounted_views
@staticmethod
async def get_shout_media(shout_slug) -> Dict[str, int]:
def get_shout_media(shout_slug) -> Dict[str, int]:
"""Получение метрики воспроизведения shout по slug."""
self = ViewedStorage
@@ -200,29 +151,23 @@ class ViewedStorage:
return self.views_by_shout.get(shout_slug, 0)
@staticmethod
async def get_topic(topic_slug) -> int:
def get_topic(topic_slug) -> int:
"""Получение суммарного значения просмотров темы."""
self = ViewedStorage
views_count = 0
for shout_slug in self.shouts_by_topic.get(topic_slug, []):
views_count += await self.get_shout(shout_slug=shout_slug)
return views_count
return sum(self.views_by_shout.get(shout_slug, 0) for shout_slug in self.shouts_by_topic.get(topic_slug, []))
@staticmethod
async def get_author(author_slug) -> int:
def get_author(author_slug) -> int:
"""Получение суммарного значения просмотров автора."""
self = ViewedStorage
views_count = 0
for shout_slug in self.shouts_by_author.get(author_slug, []):
views_count += await self.get_shout(shout_slug=shout_slug)
return views_count
return sum(self.views_by_shout.get(shout_slug, 0) for shout_slug in self.shouts_by_author.get(author_slug, []))
@staticmethod
def update_topics(shout_slug):
"""Обновление счетчиков темы по slug shout"""
self = ViewedStorage
with local_session() as session:
# Определение вспомогательной функции для избежания повторения кода
# Определение вспомогательной функции для избежа<EFBFBD><EFBFBD>ия повторения кода
def update_groups(dictionary, key, value):
dictionary[key] = list(set(dictionary.get(key, []) + [value]))

20
services/webhook.py
View File

@@ -20,7 +20,7 @@ from settings import ADMIN_SECRET, WEBHOOK_SECRET
async def check_webhook_existence():
"""
Проверяет существование вебхука для user.login события
Returns:
tuple: (bool, str, str) - существует ли вебхук, его id и endpoint если существует
"""
@@ -28,8 +28,11 @@ async def check_webhook_existence():
if not ADMIN_SECRET:
logger.error("ADMIN_SECRET is not set")
return False, None, None
headers = {"Content-Type": "application/json", "X-Authorizer-Admin-Secret": ADMIN_SECRET}
headers = {
"Content-Type": "application/json",
"X-Authorizer-Admin-Secret": ADMIN_SECRET
}
operation = "GetWebhooks"
query_name = "_webhooks"
@@ -60,14 +63,17 @@ async def create_webhook_endpoint():
"""
logger.info("create_webhook_endpoint called")
headers = {"Content-Type": "application/json", "X-Authorizer-Admin-Secret": ADMIN_SECRET}
headers = {
"Content-Type": "application/json",
"X-Authorizer-Admin-Secret": ADMIN_SECRET
}
exists, webhook_id, current_endpoint = await check_webhook_existence()
# Определяем endpoint в зависимости от окружения
host = os.environ.get("HOST", "core.dscrs.site")
host = os.environ.get('HOST', 'core.dscrs.site')
endpoint = f"https://{host}/new-author"
if exists:
# Если вебхук существует, но с другим endpoint или с модифицированным именем
if current_endpoint != endpoint or webhook_id:

22
settings.py
View File

@@ -1,29 +1,17 @@
import sys
from os import environ
MODE = "development" if "dev" in sys.argv else "production"
DEV_SERVER_PID_FILE_NAME = "dev-server.pid"
PORT = environ.get("PORT") or 8000
# storages
PORT = 8000
DB_URL = (
environ.get("DATABASE_URL", "").replace("postgres://", "postgresql://")
or environ.get("DB_URL", "").replace("postgres://", "postgresql://")
or "sqlite:///discoursio.db"
or "sqlite:///discoursio-db.sqlite3"
)
REDIS_URL = environ.get("REDIS_URL") or "redis://127.0.0.1"
# debug
AUTH_URL = environ.get("AUTH_URL") or ""
GLITCHTIP_DSN = environ.get("GLITCHTIP_DSN")
DEV_SERVER_PID_FILE_NAME = "dev-server.pid"
MODE = "development" if "dev" in sys.argv else "production"
# authorizer.dev
AUTH_URL = environ.get("AUTH_URL") or "https://auth.discours.io/graphql"
ADMIN_SECRET = environ.get("AUTH_SECRET") or "nothing"
WEBHOOK_SECRET = environ.get("WEBHOOK_SECRET") or "nothing-else"
# own auth
ONETIME_TOKEN_LIFE_SPAN = 60 * 60 * 24 * 3 # 3 days
SESSION_TOKEN_LIFE_SPAN = 60 * 60 * 24 * 30 # 30 days
JWT_ALGORITHM = "HS256"
JWT_SECRET_KEY = environ.get("JWT_SECRET") or "nothing-else-jwt-secret-matters"

25
tests/auth/conftest.py
View File

@@ -1,25 +0,0 @@
import pytest
from typing import Dict
@pytest.fixture
def oauth_settings() -> Dict[str, Dict[str, str]]:
"""Тестовые настройки OAuth"""
return {
"GOOGLE": {"id": "test_google_id", "key": "test_google_secret"},
"GITHUB": {"id": "test_github_id", "key": "test_github_secret"},
"FACEBOOK": {"id": "test_facebook_id", "key": "test_facebook_secret"},
}
@pytest.fixture
def frontend_url() -> str:
"""URL фронтенда для тестов"""
return "https://localhost:3000"
@pytest.fixture(autouse=True)
def mock_settings(monkeypatch, oauth_settings, frontend_url):
"""Подменяем настройки для тестов"""
monkeypatch.setattr("auth.oauth.OAUTH_CLIENTS", oauth_settings)
monkeypatch.setattr("auth.oauth.FRONTEND_URL", frontend_url)

222
tests/auth/test_oauth.py
View File

@@ -1,222 +0,0 @@
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from starlette.responses import JSONResponse, RedirectResponse
from auth.oauth import get_user_profile, oauth_login, oauth_callback
# Подменяем настройки для тестов
with (
patch("auth.oauth.FRONTEND_URL", "https://localhost:3000"),
patch(
"auth.oauth.OAUTH_CLIENTS",
{
"GOOGLE": {"id": "test_google_id", "key": "test_google_secret"},
"GITHUB": {"id": "test_github_id", "key": "test_github_secret"},
"FACEBOOK": {"id": "test_facebook_id", "key": "test_facebook_secret"},
},
),
):
@pytest.fixture
def mock_request():
"""Фикстура для мока запроса"""
request = MagicMock()
request.session = {}
request.path_params = {}
request.query_params = {}
return request
@pytest.fixture
def mock_oauth_client():
"""Фикстура для мока OAuth клиента"""
client = AsyncMock()
client.authorize_redirect = AsyncMock()
client.authorize_access_token = AsyncMock()
client.get = AsyncMock()
return client
@pytest.mark.asyncio
async def test_get_user_profile_google():
"""Тест получения профиля из Google"""
client = AsyncMock()
token = {
"userinfo": {
"sub": "123",
"email": "test@gmail.com",
"name": "Test User",
"picture": "https://lh3.googleusercontent.com/photo=s96",
}
}
profile = await get_user_profile("google", client, token)
assert profile["id"] == "123"
assert profile["email"] == "test@gmail.com"
assert profile["name"] == "Test User"
assert profile["picture"] == "https://lh3.googleusercontent.com/photo=s600"
@pytest.mark.asyncio
async def test_get_user_profile_github():
"""Тест получения профиля из GitHub"""
client = AsyncMock()
client.get.side_effect = [
MagicMock(
json=lambda: {
"id": 456,
"login": "testuser",
"name": "Test User",
"avatar_url": "https://github.com/avatar.jpg",
}
),
MagicMock(
json=lambda: [
{"email": "other@github.com", "primary": False},
{"email": "test@github.com", "primary": True},
]
),
]
profile = await get_user_profile("github", client, {})
assert profile["id"] == "456"
assert profile["email"] == "test@github.com"
assert profile["name"] == "Test User"
assert profile["picture"] == "https://github.com/avatar.jpg"
@pytest.mark.asyncio
async def test_get_user_profile_facebook():
"""Тест получения профиля из Facebook"""
client = AsyncMock()
client.get.return_value = MagicMock(
json=lambda: {
"id": "789",
"name": "Test User",
"email": "test@facebook.com",
"picture": {"data": {"url": "https://facebook.com/photo.jpg"}},
}
)
profile = await get_user_profile("facebook", client, {})
assert profile["id"] == "789"
assert profile["email"] == "test@facebook.com"
assert profile["name"] == "Test User"
assert profile["picture"] == "https://facebook.com/photo.jpg"
@pytest.mark.asyncio
async def test_oauth_login_success(mock_request, mock_oauth_client):
"""Тест успешного начала OAuth авторизации"""
mock_request.path_params["provider"] = "google"
# Настраиваем мок для authorize_redirect
redirect_response = RedirectResponse(url="http://example.com")
mock_oauth_client.authorize_redirect.return_value = redirect_response
with patch("auth.oauth.oauth.create_client", return_value=mock_oauth_client):
response = await oauth_login(mock_request)
assert isinstance(response, RedirectResponse)
assert mock_request.session["provider"] == "google"
assert "code_verifier" in mock_request.session
assert "state" in mock_request.session
mock_oauth_client.authorize_redirect.assert_called_once()
@pytest.mark.asyncio
async def test_oauth_login_invalid_provider(mock_request):
"""Тест с неправильным провайдером"""
mock_request.path_params["provider"] = "invalid"
response = await oauth_login(mock_request)
assert isinstance(response, JSONResponse)
assert response.status_code == 400
assert "Invalid provider" in response.body.decode()
@pytest.mark.asyncio
async def test_oauth_callback_success(mock_request, mock_oauth_client):
"""Тест успешного OAuth callback"""
mock_request.session = {
"provider": "google",
"code_verifier": "test_verifier",
"state": "test_state",
}
mock_request.query_params["state"] = "test_state"
mock_oauth_client.authorize_access_token.return_value = {
"userinfo": {"sub": "123", "email": "test@gmail.com", "name": "Test User"}
}
with (
patch("auth.oauth.oauth.create_client", return_value=mock_oauth_client),
patch("auth.oauth.local_session") as mock_session,
patch("auth.oauth.TokenStorage.create_session", return_value="test_token"),
):
# Мокаем сессию базы данных
session = MagicMock()
session.query.return_value.filter.return_value.first.return_value = None
mock_session.return_value.__enter__.return_value = session
response = await oauth_callback(mock_request)
assert isinstance(response, RedirectResponse)
assert response.status_code == 307
assert "auth/success" in response.headers["location"]
# Проверяем cookie
cookies = response.headers.getlist("set-cookie")
assert any("session_token=test_token" in cookie for cookie in cookies)
assert any("httponly" in cookie.lower() for cookie in cookies)
assert any("secure" in cookie.lower() for cookie in cookies)
# Проверяем очистку сессии
assert "code_verifier" not in mock_request.session
assert "provider" not in mock_request.session
assert "state" not in mock_request.session
@pytest.mark.asyncio
async def test_oauth_callback_invalid_state(mock_request):
"""Тест с неправильным state параметром"""
mock_request.session = {"provider": "google", "state": "correct_state"}
mock_request.query_params["state"] = "wrong_state"
response = await oauth_callback(mock_request)
assert isinstance(response, JSONResponse)
assert response.status_code == 400
assert "Invalid state" in response.body.decode()
@pytest.mark.asyncio
async def test_oauth_callback_existing_user(mock_request, mock_oauth_client):
"""Тест OAuth callback с существующим пользователем"""
mock_request.session = {
"provider": "google",
"code_verifier": "test_verifier",
"state": "test_state",
}
mock_request.query_params["state"] = "test_state"
mock_oauth_client.authorize_access_token.return_value = {
"userinfo": {"sub": "123", "email": "test@gmail.com", "name": "Test User"}
}
with (
patch("auth.oauth.oauth.create_client", return_value=mock_oauth_client),
patch("auth.oauth.local_session") as mock_session,
patch("auth.oauth.TokenStorage.create_session", return_value="test_token"),
):
# Мокаем существующего пользователя
existing_user = MagicMock()
session = MagicMock()
session.query.return_value.filter.return_value.first.return_value = existing_user
mock_session.return_value.__enter__.return_value = session
response = await oauth_callback(mock_request)
assert isinstance(response, RedirectResponse)
assert response.status_code == 307
# Проверяем обновление существующего пользователя
assert existing_user.name == "Test User"
assert existing_user.oauth == "google:123"
assert existing_user.email_verified is True

9
tests/auth/test_settings.py
View File

@@ -1,9 +0,0 @@
"""Тестовые настройки для OAuth"""
FRONTEND_URL = "https://localhost:3000"
OAUTH_CLIENTS = {
"GOOGLE": {"id": "test_google_id", "key": "test_google_secret"},
"GITHUB": {"id": "test_github_id", "key": "test_github_secret"},
"FACEBOOK": {"id": "test_facebook_id", "key": "test_facebook_secret"},
}

48
tests/conftest.py
View File

@@ -1,48 +0,0 @@
import asyncio
import pytest
from services.redis import redis
from tests.test_config import get_test_client
@pytest.fixture(scope="session")
def event_loop():
"""Create an instance of the default event loop for the test session."""
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="session")
def test_app():
"""Create a test client and session factory."""
client, SessionLocal = get_test_client()
return client, SessionLocal
@pytest.fixture
def db_session(test_app):
"""Create a new database session for a test."""
_, SessionLocal = test_app
session = SessionLocal()
yield session
session.rollback()
session.close()
@pytest.fixture
def test_client(test_app):
"""Get the test client."""
client, _ = test_app
return client
@pytest.fixture
async def redis_client():
"""Create a test Redis client."""
await redis.connect()
await redis.flushall() # Очищаем Redis перед каждым тестом
yield redis
await redis.flushall() # Очищаем после теста
await redis.disconnect()

67
tests/test_config.py
View File

@@ -1,67 +0,0 @@
"""
Конфигурация для тестов
"""
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.testclient import TestClient
# Используем in-memory SQLite для тестов
TEST_DB_URL = "sqlite:///:memory:"
class DatabaseMiddleware(BaseHTTPMiddleware):
"""Middleware для внедрения сессии БД"""
def __init__(self, app, session_maker):
super().__init__(app)
self.session_maker = session_maker
async def dispatch(self, request, call_next):
session = self.session_maker()
request.state.db = session
try:
response = await call_next(request)
finally:
session.close()
return response
def create_test_app():
"""Create a test Starlette application."""
from services.db import Base
# Создаем движок и таблицы
engine = create_engine(
TEST_DB_URL,
connect_args={"check_same_thread": False},
poolclass=StaticPool,
echo=False,
)
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)
# Создаем фабрику сессий
SessionLocal = sessionmaker(bind=engine)
# Создаем middleware для сессий
middleware = [Middleware(DatabaseMiddleware, session_maker=SessionLocal)]
# Создаем тестовое приложение
app = Starlette(
debug=True,
middleware=middleware,
routes=[], # Здесь можно добавить тестовые маршруты если нужно
)
return app, SessionLocal
def get_test_client():
"""Get a test client with initialized database."""
app, SessionLocal = create_test_app()
return TestClient(app), SessionLocal

94
tests/test_drafts.py
View File

@@ -1,94 +0,0 @@
import pytest
from orm.author import Author
from orm.shout import Shout
@pytest.fixture
def test_author(db_session):
"""Create a test author."""
author = Author(name="Test Author", slug="test-author", user="test-user-id")
db_session.add(author)
db_session.commit()
return author
@pytest.fixture
def test_shout(db_session):
"""Create test shout with required fields."""
author = Author(name="Test Author", slug="test-author", user="test-user-id")
db_session.add(author)
db_session.flush()
shout = Shout(
title="Test Shout",
slug="test-shout",
created_by=author.id, # Обязательное поле
body="Test body",
layout="article",
lang="ru",
)
db_session.add(shout)
db_session.commit()
return shout
@pytest.mark.asyncio
async def test_create_shout(test_client, db_session, test_author):
"""Test creating a new shout."""
response = test_client.post(
"/",
json={
"query": """
mutation CreateDraft($draft_input: DraftInput!) {
create_draft(draft_input: $draft_input) {
error
draft {
id
title
body
}
}
}
""",
"variables": {
"input": {
"title": "Test Shout",
"body": "This is a test shout",
}
},
},
)
assert response.status_code == 200
data = response.json()
assert "errors" not in data
assert data["data"]["create_draft"]["draft"]["title"] == "Test Shout"
@pytest.mark.asyncio
async def test_load_drafts(test_client, db_session):
"""Test retrieving a shout."""
response = test_client.post(
"/",
json={
"query": """
query {
load_drafts {
error
drafts {
id
title
body
}
}
}
""",
"variables": {"slug": "test-shout"},
},
)
assert response.status_code == 200
data = response.json()
assert "errors" not in data
assert data["data"]["load_drafts"]["drafts"] == []

68
tests/test_reactions.py
View File

@@ -1,68 +0,0 @@
from datetime import datetime
import pytest
from orm.author import Author
from orm.reaction import ReactionKind
from orm.shout import Shout
@pytest.fixture
def test_setup(db_session):
"""Set up test data."""
now = int(datetime.now().timestamp())
author = Author(name="Test Author", slug="test-author", user="test-user-id")
db_session.add(author)
db_session.flush()
shout = Shout(
title="Test Shout",
slug="test-shout",
created_by=author.id,
body="This is a test shout",
layout="article",
lang="ru",
community=1,
created_at=now,
updated_at=now,
)
db_session.add_all([author, shout])
db_session.commit()
return {"author": author, "shout": shout}
@pytest.mark.asyncio
async def test_create_reaction(test_client, db_session, test_setup):
"""Test creating a reaction on a shout."""
response = test_client.post(
"/",
json={
"query": """
mutation CreateReaction($reaction: ReactionInput!) {
create_reaction(reaction: $reaction) {
error
reaction {
id
kind
body
created_by {
name
}
}
}
}
""",
"variables": {
"reaction": {
"shout": test_setup["shout"].id,
"kind": ReactionKind.LIKE.value,
"body": "Great post!",
}
},
},
)
assert response.status_code == 200
data = response.json()
assert "error" not in data
assert data["data"]["create_reaction"]["reaction"]["kind"] == ReactionKind.LIKE.value

85
tests/test_shouts.py
View File

@@ -1,85 +0,0 @@
from datetime import datetime
import pytest
from orm.author import Author
from orm.shout import Shout
@pytest.fixture
def test_shout(db_session):
"""Create test shout with required fields."""
now = int(datetime.now().timestamp())
author = Author(name="Test Author", slug="test-author", user="test-user-id")
db_session.add(author)
db_session.flush()
now = int(datetime.now().timestamp())
shout = Shout(
title="Test Shout",
slug="test-shout",
created_by=author.id,
body="Test body",
layout="article",
lang="ru",
community=1,
created_at=now,
updated_at=now,
)
db_session.add(shout)
db_session.commit()
return shout
@pytest.mark.asyncio
async def test_get_shout(test_client, db_session):
"""Test retrieving a shout."""
# Создаем автора
author = Author(name="Test Author", slug="test-author", user="test-user-id")
db_session.add(author)
db_session.flush()
now = int(datetime.now().timestamp())
# Создаем публикацию со всеми обязательными полями
shout = Shout(
title="Test Shout",
body="This is a test shout",
slug="test-shout",
created_by=author.id,
layout="article",
lang="ru",
community=1,
created_at=now,
updated_at=now,
)
db_session.add(shout)
db_session.commit()
response = test_client.post(
"/",
json={
"query": """
query GetShout($slug: String!) {
get_shout(slug: $slug) {
id
title
body
created_at
updated_at
created_by {
id
name
slug
}
}
}
""",
"variables": {"slug": "test-shout"},
},
)
data = response.json()
assert response.status_code == 200
assert "errors" not in data
assert data["data"]["get_shout"]["title"] == "Test Shout"

1
utils/__init__.py
View File

@@ -1 +0,0 @@

23
utils/encoders.py
View File

@@ -1,28 +1,9 @@
import json
from decimal import Decimal
from json import JSONEncoder
class CustomJSONEncoder(JSONEncoder):
"""
Расширенный JSON энкодер с поддержкой сериализации объектов SQLAlchemy.
Примеры:
>>> import json
>>> from decimal import Decimal
>>> from orm.topic import Topic
>>> json.dumps(Decimal("10.50"), cls=CustomJSONEncoder)
'"10.50"'
>>> topic = Topic(id=1, slug="test")
>>> json.dumps(topic, cls=CustomJSONEncoder)
'{"id": 1, "slug": "test", ...}'
"""
class CustomJSONEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return str(obj)
# Проверяем, есть ли у объекта метод dict() (как у моделей SQLAlchemy)
if hasattr(obj, "dict") and callable(obj.dict):
return obj.dict()
return super().default(obj)

38
utils/html_wrapper.py
View File

@@ -1,38 +0,0 @@
"""
Модуль для обработки HTML-фрагментов
"""
def wrap_html_fragment(fragment: str) -> str:
"""
Оборачивает HTML-фрагмент в полную HTML-структуру для корректной обработки.
Args:
fragment: HTML-фрагмент для обработки
Returns:
str: Полный HTML-документ
Example:
>>> wrap_html_fragment("<p>Текст параграфа</p>")
'<!DOCTYPE html><html><head><meta charset="utf-8"></head><body><p>Текст параграфа</p></body></html>'
"""
if not fragment or not fragment.strip():
return fragment
# Проверяем, является ли контент полным HTML-документом
is_full_html = fragment.strip().startswith('<!DOCTYPE') or fragment.strip().startswith('<html')
# Если это фрагмент, оборачиваем его в полный HTML-документ
if not is_full_html:
return f"""<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title></title>
</head>
<body>
{fragment}
</body>
</html>"""
return fragment

59
utils/logger.py
View File

@@ -1,38 +1,15 @@
import logging
from pathlib import Path
import colorlog
_lib_path = Path(__file__).parents[1]
_leng_path = len(_lib_path.as_posix())
def filter(record: logging.LogRecord):
# Define `package` attribute with the relative path.
record.package = record.pathname[_leng_path + 1 :].replace(".py", "")
record.emoji = (
"🔍"
if record.levelno == logging.DEBUG
else "ℹ︎"
if record.levelno == logging.INFO
else "🚧"
if record.levelno == logging.WARNING
else "❌"
if record.levelno == logging.ERROR
else "🧨"
if record.levelno == logging.CRITICAL
else ""
)
return record
# Define the color scheme
color_scheme = {
"DEBUG": "light_black",
"DEBUG": "cyan",
"INFO": "green",
"WARNING": "yellow",
"ERROR": "red",
"CRITICAL": "red,bg_white",
"DEFAULT": "white",
}
# Define secondary log colors
@@ -40,12 +17,12 @@ secondary_colors = {
"log_name": {"DEBUG": "blue"},
"asctime": {"DEBUG": "cyan"},
"process": {"DEBUG": "purple"},
"module": {"DEBUG": "light_black,bg_blue"},
"funcName": {"DEBUG": "light_white,bg_blue"}, # Add this line
"module": {"DEBUG": "cyan,bg_blue"},
"funcName": {"DEBUG": "light_white,bg_blue"},
}
# Define the log format string
fmt_string = "%(emoji)s%(log_color)s%(package)s.%(funcName)s%(reset)s %(white)s%(message)s"
fmt_string = "%(log_color)s%(levelname)s: %(log_color)s[%(module)s.%(funcName)s]%(reset)s %(white)s%(message)s"
# Define formatting configuration
fmt_config = {
@@ -63,10 +40,6 @@ class MultilineColoredFormatter(colorlog.ColoredFormatter):
self.secondary_log_colors = kwargs.pop("secondary_log_colors", {})
def format(self, record):
# Add default emoji if not present
if not hasattr(record, "emoji"):
record = filter(record)
message = record.getMessage()
if "\n" in message:
lines = message.split("\n")
@@ -88,24 +61,8 @@ formatter = MultilineColoredFormatter(fmt_string, **fmt_config)
stream = logging.StreamHandler()
stream.setFormatter(formatter)
def get_colorful_logger(name="main"):
# Create and configure the logger
logger = logging.getLogger(name)
logger.setLevel(logging.DEBUG)
logger.addHandler(stream)
logger.addFilter(filter)
return logger
# Set up the root logger with the same formatting
root_logger = logging.getLogger()
root_logger.setLevel(logging.DEBUG)
root_logger.addHandler(stream)
root_logger.addFilter(filter)
ignore_logs = ["_trace", "httpx", "_client", "atrace", "aiohttp", "_client"]
for lgr in ignore_logs:
loggr = logging.getLogger(lgr)
loggr.setLevel(logging.INFO)
if not root_logger.hasHandlers():
root_logger.setLevel(logging.DEBUG)
root_logger.addHandler(stream)