from cachetools import TTLCache, cached from flask import current_app import datetime from sqlalchemy.sql.expression import literal from sqlalchemy import func, or_, desc, and_, distinct, cast, Time from backend import db from backend.models.dtos.project_dto import ProjectFavoritesDTO, ProjectSearchResultsDTO from backend.models.dtos.user_dto import ( UserDTO, UserOSMDTO, UserFilterDTO, UserSearchQuery, UserSearchDTO, UserStatsDTO, UserContributionDTO, UserRegisterEmailDTO, UserCountryContributed, UserCountriesContributed, ) from backend.models.dtos.interests_dto import InterestsListDTO, InterestDTO from backend.models.postgis.interests import Interest, project_interests from backend.models.postgis.message import Message from backend.models.postgis.project import Project from backend.models.postgis.user import User, UserRole, MappingLevel, UserEmail from backend.models.postgis.task import TaskHistory, TaskAction, Task from backend.models.dtos.user_dto import UserTaskDTOs from backend.models.dtos.stats_dto import Pagination from backend.models.postgis.statuses import TaskStatus, ProjectStatus from backend.models.postgis.utils import NotFound from backend.services.users.osm_service import OSMService, OSMServiceError from backend.services.messaging.smtp_service import SMTPService from backend.services.messaging.template_service import ( get_txt_template, template_var_replacing, ) user_filter_cache = TTLCache(maxsize=1024, ttl=600) class UserServiceError(Exception): """ Custom Exception to notify callers an error occurred when in the User Service """ def __init__(self, message): if current_app: current_app.logger.debug(message) class UserService: @staticmethod def get_user_by_id(user_id: int) -> User: user = User.get_by_id(user_id) if user is None: raise NotFound() return user @staticmethod def get_user_by_username(username: str) -> User: user = User.get_by_username(username) if user is None: raise NotFound() return user @staticmethod def get_contributions_by_day(user_id: int): # Validate that user exists. stats = ( TaskHistory.query.with_entities( func.DATE(TaskHistory.action_date).label("day"), func.count(TaskHistory.action).label("cnt"), ) .filter(TaskHistory.user_id == user_id) .filter(TaskHistory.action == TaskAction.STATE_CHANGE.name) .filter( func.DATE(TaskHistory.action_date) > datetime.date.today() - datetime.timedelta(days=365) ) .group_by("day") .order_by(desc("day")) ) contributions = [ UserContributionDTO(dict(date=str(s[0]), count=s[1])) for s in stats ] return contributions @staticmethod def get_project_managers() -> User: users = User.query.filter(User.role == 2).all() if users is None: raise NotFound() return users @staticmethod def get_general_admins() -> User: users = User.query.filter(User.role == 1).all() if users is None: raise NotFound() return users @staticmethod def update_user(user_id: int, osm_username: str, picture_url: str) -> User: user = UserService.get_user_by_id(user_id) if user.username != osm_username: user.update_username(osm_username) if picture_url is not None and user.picture_url != picture_url: user.update_picture_url(picture_url) return user @staticmethod def get_projects_favorited(user_id: int) -> ProjectFavoritesDTO: user = UserService.get_user_by_id(user_id) projects_dto = [f.as_dto_for_admin(f.id) for f in user.favorites] fav_dto = ProjectFavoritesDTO() fav_dto.favorited_projects = projects_dto return fav_dto @staticmethod def get_projects_mapped(user_id: int): user = UserService.get_user_by_id(user_id) projects_mapped = user.projects_mapped # Return empty list if the user has no projects_mapped. if projects_mapped is None: return [] return projects_mapped @staticmethod def register_user(osm_id, username, changeset_count, picture_url, email): """ Creates user in DB :param osm_id: Unique OSM user id :param username: OSM Username :param changeset_count: OSM changeset count """ new_user = User() new_user.id = osm_id new_user.username = username if picture_url is not None: new_user.picture_url = picture_url intermediate_level = current_app.config["MAPPER_LEVEL_INTERMEDIATE"] advanced_level = current_app.config["MAPPER_LEVEL_ADVANCED"] if changeset_count > advanced_level: new_user.mapping_level = MappingLevel.ADVANCED.value elif intermediate_level < changeset_count < advanced_level: new_user.mapping_level = MappingLevel.INTERMEDIATE.value else: new_user.mapping_level = MappingLevel.BEGINNER.value if email is not None: new_user.email_address = email new_user.create() return new_user @staticmethod def get_user_dto_by_username( requested_username: str, logged_in_user_id: int ) -> UserDTO: """Gets user DTO for supplied username """ requested_user = UserService.get_user_by_username(requested_username) logged_in_user = UserService.get_user_by_id(logged_in_user_id) UserService.check_and_update_mapper_level(requested_user.id) return requested_user.as_dto(logged_in_user.username) @staticmethod def get_user_dto_by_id(user: int, request_user: int) -> UserDTO: """Gets user DTO for supplied user id """ user = UserService.get_user_by_id(user) if request_user: request_username = UserService.get_user_by_id(request_user).username return user.as_dto(request_username) return user.as_dto() @staticmethod def get_interests_stats(user_id): # Get all projects that the user has contributed. stmt = ( TaskHistory.query.with_entities(TaskHistory.project_id) .distinct() .filter(TaskHistory.user_id == user_id) ) interests = ( Interest.query.with_entities( Interest.id, Interest.name, func.count(distinct(project_interests.c.project_id)).label( "count_projects" ), ) .join( project_interests, and_( Interest.id == project_interests.c.interest_id, project_interests.c.project_id.in_(stmt), ), ) .group_by(Interest.id) .order_by(desc("count_projects")) .all() ) interests_dto = [ InterestDTO(dict(id=i.id, name=i.name, count_projects=i.count_projects)) for i in interests ] return interests_dto @staticmethod def get_tasks_dto( user_id: int, start_date: datetime.datetime = None, end_date: datetime.datetime = None, task_status: str = None, project_status: str = None, project_id: int = None, page=1, page_size=10, sort_by: str = None, ) -> UserTaskDTOs: base_query = ( TaskHistory.query.with_entities( TaskHistory.project_id.label("project_id"), TaskHistory.task_id.label("task_id"), func.max(TaskHistory.action_date).label("max"), ) .filter(TaskHistory.user_id == user_id) .group_by(TaskHistory.task_id, TaskHistory.project_id) ) if task_status: base_query = base_query.filter( TaskHistory.action_text == TaskStatus[task_status.upper()].name ) if start_date: base_query = base_query.filter(TaskHistory.action_date >= start_date) if end_date: base_query = base_query.filter(TaskHistory.action_date <= end_date) if sort_by == "action_date": base_query = base_query.order_by(func.max(TaskHistory.action_date)) elif sort_by == "-action_date": base_query = base_query.order_by(desc(func.max(TaskHistory.action_date))) user_task_dtos = UserTaskDTOs() task_id_list = base_query.subquery() comments_query = ( TaskHistory.query.with_entities( TaskHistory.project_id, TaskHistory.task_id, func.count(TaskHistory.action).label("count"), ) .filter(TaskHistory.action == "COMMENT") .group_by(TaskHistory.task_id, TaskHistory.project_id) ).subquery() sq = ( db.session.query( func.coalesce(comments_query.c.count, 0).label("comments"), task_id_list ) .select_from(task_id_list) .outerjoin( comments_query, (comments_query.c.task_id == task_id_list.c.task_id) & (comments_query.c.project_id == task_id_list.c.project_id), ) .subquery() ) tasks = Task.query.join( sq, and_( Task.id == sq.c.task_id, Task.project_id == sq.c.project_id, ), ) tasks = tasks.add_columns("max", "comments") if project_status: tasks = tasks.filter( Task.project_id == Project.id, Project.status == ProjectStatus[project_status.upper()].value, ) if project_id: tasks = tasks.filter_by(project_id=project_id) results = tasks.paginate(page, page_size, True) task_list = [] for task, action_date, comments in results.items: task_list.append(task.as_dto(last_updated=action_date, comments=comments)) user_task_dtos.user_tasks = task_list user_task_dtos.pagination = Pagination(results) return user_task_dtos @staticmethod def get_detailed_stats(username: str): user = UserService.get_user_by_username(username) stats_dto = UserStatsDTO() actions = [ TaskStatus.VALIDATED.name, TaskStatus.INVALIDATED.name, TaskStatus.MAPPED.name, ] actions_table = ( db.session.query(literal(TaskStatus.VALIDATED.name).label("action_text")) .union( db.session.query( literal(TaskStatus.INVALIDATED.name).label("action_text") ), db.session.query(literal(TaskStatus.MAPPED.name).label("action_text")), ) .subquery() .alias("actions_table") ) # Get only rows with the given actions. filtered_actions = ( TaskHistory.query.with_entities( TaskHistory.user_id, TaskHistory.project_id, TaskHistory.task_id, TaskHistory.action_text, ) .filter(TaskHistory.action_text.in_(actions)) .subquery() .alias("filtered_actions") ) user_tasks = ( db.session.query(filtered_actions) .filter(filtered_actions.c.user_id == user.id) .subquery() .alias("user_tasks") ) others_tasks = ( db.session.query(filtered_actions) .filter(filtered_actions.c.user_id != user.id) .filter(filtered_actions.c.task_id == user_tasks.c.task_id) .filter(filtered_actions.c.project_id == user_tasks.c.project_id) .filter(filtered_actions.c.action_text != TaskStatus.MAPPED.name) .subquery() .alias("others_tasks") ) user_stats = ( db.session.query( actions_table.c.action_text, func.count(user_tasks.c.action_text) ) .outerjoin( user_tasks, actions_table.c.action_text == user_tasks.c.action_text ) .group_by(actions_table.c.action_text) ) others_stats = ( db.session.query( func.concat(actions_table.c.action_text, "_BY_OTHERS"), func.count(others_tasks.c.action_text), ) .outerjoin( others_tasks, actions_table.c.action_text == others_tasks.c.action_text ) .group_by(actions_table.c.action_text) ) res = user_stats.union(others_stats).all() results = {key: value for key, value in res} projects_mapped = UserService.get_projects_mapped(user.id) stats_dto.tasks_mapped = results["MAPPED"] stats_dto.tasks_validated = results["VALIDATED"] stats_dto.tasks_invalidated = results["INVALIDATED"] stats_dto.tasks_validated_by_others = results["VALIDATED_BY_OTHERS"] stats_dto.tasks_invalidated_by_others = results["INVALIDATED_BY_OTHERS"] stats_dto.projects_mapped = len(projects_mapped) stats_dto.countries_contributed = UserService.get_countries_contributed(user.id) stats_dto.contributions_by_day = UserService.get_contributions_by_day(user.id) stats_dto.total_time_spent = 0 stats_dto.time_spent_mapping = 0 stats_dto.time_spent_validating = 0 query = ( TaskHistory.query.with_entities( func.date_trunc("minute", TaskHistory.action_date).label("trn"), func.max(TaskHistory.action_text).label("tm"), ) .filter(TaskHistory.user_id == user.id) .filter(TaskHistory.action == "LOCKED_FOR_VALIDATION") .group_by("trn") .subquery() ) total_validation_time = db.session.query( func.sum(cast(func.to_timestamp(query.c.tm, "HH24:MI:SS"), Time)) ).scalar() if total_validation_time: stats_dto.time_spent_validating = total_validation_time.total_seconds() stats_dto.total_time_spent += stats_dto.time_spent_validating total_mapping_time = ( db.session.query( func.sum( cast(func.to_timestamp(TaskHistory.action_text, "HH24:MI:SS"), Time) ) ) .filter( or_( TaskHistory.action == TaskAction.LOCKED_FOR_MAPPING.name, TaskHistory.action == TaskAction.AUTO_UNLOCKED_FOR_MAPPING.name, ) ) .filter(TaskHistory.user_id == user.id) .scalar() ) if total_mapping_time: stats_dto.time_spent_mapping = total_mapping_time.total_seconds() stats_dto.total_time_spent += stats_dto.time_spent_mapping stats_dto.contributions_interest = UserService.get_interests_stats(user.id) return stats_dto @staticmethod def update_user_details(user_id: int, user_dto: UserDTO) -> dict: """Update user with info supplied by user, if they add or change their email address a verification mail will be sent""" user = UserService.get_user_by_id(user_id) verification_email_sent = False if ( user_dto.email_address and user.email_address != user_dto.email_address.lower() ): # Send user verification email if they are adding or changing their email address SMTPService.send_verification_email( user_dto.email_address.lower(), user.username ) user.set_email_verified_status(is_verified=False) verification_email_sent = True user.update(user_dto) user_email = UserEmail.query.filter( UserEmail.email == user_dto.email_address ).one_or_none() if user_email is not None: user_email.delete() return dict(verificationEmailSent=verification_email_sent) @staticmethod def get_all_users(query: UserSearchQuery) -> UserSearchDTO: """ Gets paginated list of users """ return User.get_all_users(query) @staticmethod @cached(user_filter_cache) def filter_users(username: str, project_id: int, page: int) -> UserFilterDTO: """ Gets paginated list of users, filtered by username, for autocomplete """ return User.filter_users(username, project_id, page) @staticmethod def is_user_an_admin(user_id: int) -> bool: """ Is the user an admin """ user = UserService.get_user_by_id(user_id) if UserRole(user.role) == UserRole.ADMIN: return True return False @staticmethod def is_user_the_project_author(user_id: int, author_id: int) -> bool: """ Is user the author of the project """ return user_id == author_id @staticmethod def get_mapping_level(user_id: int): """ Gets mapping level user is at""" user = UserService.get_user_by_id(user_id) return MappingLevel(user.mapping_level) @staticmethod def is_user_validator(user_id: int) -> bool: """ Determines if user is a validator """ user = UserService.get_user_by_id(user_id) if UserRole(user.role) in [ UserRole.ADMIN, ]: return True return False @staticmethod def is_user_blocked(user_id: int) -> bool: """ Determines if a user is blocked """ user = UserService.get_user_by_id(user_id) if UserRole(user.role) == UserRole.READ_ONLY: return True return False @staticmethod def get_countries_contributed(user_id: int): query = ( TaskHistory.query.with_entities( func.unnest(Project.country).label("country"), TaskHistory.action_text, func.count(TaskHistory.action_text).label("count"), ) .filter(TaskHistory.user_id == user_id) .filter( TaskHistory.action_text.in_( [ TaskStatus.MAPPED.name, TaskStatus.BADIMAGERY.name, TaskStatus.VALIDATED.name, ] ) ) .group_by("country", TaskHistory.action_text) .outerjoin(Project, Project.id == TaskHistory.project_id) .all() ) countries = list(set([q.country for q in query])) result = [] for country in countries: values = [q for q in query if q.country == country] # Filter element to sum mapped values. mapped = sum( [ v.count for v in values if v.action_text in [TaskStatus.MAPPED.name, TaskStatus.BADIMAGERY.name] ] ) validated = sum( [v.count for v in values if v.action_text == TaskStatus.VALIDATED.name] ) dto = UserCountryContributed( dict( name=country, mapped=mapped, validated=validated, total=mapped + validated, ) ) result.append(dto) # Order by total result = sorted(result, reverse=True, key=lambda i: i.total) countries_dto = UserCountriesContributed() countries_dto.countries_contributed = result countries_dto.total = len(result) return countries_dto @staticmethod def upsert_mapped_projects(user_id: int, project_id: int): """ Add project to mapped projects if it doesn't exist, otherwise return """ User.upsert_mapped_projects(user_id, project_id) @staticmethod def get_mapped_projects(user_name: str, preferred_locale: str): """ Gets all projects a user has mapped or validated on """ user = UserService.get_user_by_username(user_name) return User.get_mapped_projects(user.id, preferred_locale) @staticmethod def get_recommended_projects(user_name: str, preferred_locale: str): """ Gets all projects a user has mapped or validated on """ from backend.services.project_search_service import ProjectSearchService limit = 20 user = ( User.query.with_entities(User.id, User.mapping_level) .filter(User.username == user_name) .one_or_none() ) if user is None: raise NotFound() # Get all projects that the user has contributed sq = ( TaskHistory.query.with_entities(TaskHistory.project_id.label("project_id")) .distinct(TaskHistory.project_id) .filter(TaskHistory.user_id == user.id) .subquery() ) # Get all campaigns for all contributed projects. campaign_tags = ( Project.query.with_entities(Project.campaign.label("tag")) .filter(or_(Project.author_id == user.id, Project.id == sq.c.project_id)) .subquery() ) # Get projects with given campaign tags but without user contributions. query = ProjectSearchService.create_search_query() projs = ( query.filter(Project.campaign.any(campaign_tags.c.tag)).limit(limit).all() ) # Get only user mapping level projects. len_projs = len(projs) if len_projs < limit: remaining_projs = ( query.filter(Project.mapper_level == user.mapping_level) .limit(limit - len_projs) .all() ) projs.extend(remaining_projs) dto = ProjectSearchResultsDTO() # Get all total contributions for each paginated project. contrib_counts = ProjectSearchService.get_total_contributions(projs) zip_items = zip(projs, contrib_counts) dto.results = [ ProjectSearchService.create_result_dto(p, "en", t) for p, t in zip_items ] return dto @staticmethod def add_role_to_user(admin_user_id: int, username: str, role: str): """ Add role to user :param admin_user_id: ID of admin attempting to add the role :param username: Username of user the role should be added to :param role: The requested role :raises UserServiceError """ try: requested_role = UserRole[role.upper()] except KeyError: raise UserServiceError( f"Unknown role {role} accepted values are ADMIN, PROJECT_MANAGER, VALIDATOR" ) admin = UserService.get_user_by_id(admin_user_id) admin_role = UserRole(admin.role) if admin_role != UserRole.ADMIN and requested_role == UserRole.ADMIN: raise UserServiceError("You must be an Admin to assign Admin role") user = UserService.get_user_by_username(username) user.set_user_role(requested_role) @staticmethod def set_user_mapping_level(username: str, level: str) -> User: """ Sets the users mapping level :raises: UserServiceError """ try: requested_level = MappingLevel[level.upper()] except KeyError: raise UserServiceError( f"Unknown role {level} accepted values are BEGINNER, INTERMEDIATE, ADVANCED" ) user = UserService.get_user_by_username(username) user.set_mapping_level(requested_level) return user @staticmethod def set_user_is_expert(user_id: int, is_expert: bool) -> User: """ Enabled or disables expert mode for the user :raises: UserServiceError """ user = UserService.get_user_by_id(user_id) user.set_is_expert(is_expert) return user @staticmethod def accept_license_terms(user_id: int, license_id: int): """ Saves the fact user has accepted license terms """ user = UserService.get_user_by_id(user_id) user.accept_license_terms(license_id) @staticmethod def has_user_accepted_license(user_id: int, license_id: int): """ Checks if user has accepted specified license """ user = UserService.get_user_by_id(user_id) return user.has_user_accepted_licence(license_id) @staticmethod def get_osm_details_for_user(username: str) -> UserOSMDTO: """ Gets OSM details for the user from OSM API :param username: username in scope :raises UserServiceError, NotFound """ user = UserService.get_user_by_username(username) osm_dto = OSMService.get_osm_details_for_user(user.id) return osm_dto @staticmethod def check_and_update_mapper_level(user_id: int): """ Check users mapping level and update if they have crossed threshold """ user = UserService.get_user_by_id(user_id) user_level = MappingLevel(user.mapping_level) if user_level == MappingLevel.ADVANCED: return # User has achieved highest level, so no need to do further checking intermediate_level = current_app.config["MAPPER_LEVEL_INTERMEDIATE"] advanced_level = current_app.config["MAPPER_LEVEL_ADVANCED"] try: osm_details = OSMService.get_osm_details_for_user(user_id) if ( osm_details.changeset_count > advanced_level and user.mapping_level != MappingLevel.ADVANCED.value ): user.mapping_level = MappingLevel.ADVANCED.value UserService.notify_level_upgrade(user_id, user.username, "ADVANCED") elif ( intermediate_level < osm_details.changeset_count < advanced_level and user.mapping_level != MappingLevel.INTERMEDIATE.value ): user.mapping_level = MappingLevel.INTERMEDIATE.value UserService.notify_level_upgrade(user_id, user.username, "INTERMEDIATE") except OSMServiceError: # Swallow exception as we don't want to blow up the server for this current_app.logger.error("Error attempting to update mapper level") return user.save() @staticmethod def notify_level_upgrade(user_id: int, username: str, level: str): text_template = get_txt_template("level_upgrade_message_en.txt") replace_list = [ ["[USERNAME]", username], ["[LEVEL]", level], ["[ORG_CODE]", current_app.config["ORG_CODE"]], ] text_template = template_var_replacing(text_template, replace_list) level_upgrade_message = Message() level_upgrade_message.to_user_id = user_id level_upgrade_message.subject = "Mapper level upgrade" level_upgrade_message.message = text_template level_upgrade_message.save() @staticmethod def refresh_mapper_level() -> int: """ Helper function to run thru all users in the DB and update their mapper level """ users = User.get_all_users_not_paginated() users_updated = 1 total_users = len(users) for user in users: UserService.check_and_update_mapper_level(user.id) if users_updated % 50 == 0: print(f"{users_updated} users updated of {total_users}") users_updated += 1 return users_updated @staticmethod def register_user_with_email(user_dto: UserRegisterEmailDTO): # Validate that user is not within the general users table. user = User.query.filter(User.email_address == user_dto.email).one_or_none() if user is not None: details_msg = f"Email address {user_dto.email} already exists" raise ValueError(details_msg) user = UserEmail.query.filter(UserEmail.email == user_dto.email).one_or_none() if user is None: user = UserEmail(email=user_dto.email) user.create() return user @staticmethod def get_interests(user: User) -> InterestsListDTO: dto = InterestsListDTO() for interest in Interest.query.all(): int_dto = interest.as_dto() if interest in user.interests: int_dto.user_selected = True dto.interests.append(int_dto) return dto