# { # "Seq": [ # { "Depends": "py-lib-genlayermodelwrappers:test" }, # { "Depends": "py-genlayer:test" } # ] # } import numpy as np import json import re import typing import urllib.parse from dataclasses import dataclass from datetime import datetime, timedelta, timezone from itertools import groupby from math import log10 import genlayer.py.eth as genvm_eth import genlayer.std.advanced as advanced import genlayermodelwrappers from genlayer import * # TODO: with new GenVM, use deterministic principles where possible AT_MOST_2_DIFFERENCE_PRINCIPLE = "The total score should differ by at most 2" def to_atto(value: float) -> int: return int(value * 1e18) def from_atto(value: int) -> float: return value / 1e18 MAX_SCORE_DIFFERENCE = 2 MINIMUM_CATEGORY_SCORE_ATTO = to_atto(2) @allow_storage @dataclass class VectorStoreValue: text: str id: str @allow_storage @dataclass class AnalysisRecord: atto_score: u256 analysis: str category: str atto_max_score: u256 @allow_storage @dataclass class Submission: mission_id: str user: str tweet_id: str atto_atemporal_points: u256 atto_temporal_points: u256 period_index: u256 timestamp: str analysis: DynArray[AnalysisRecord] @allow_storage @dataclass class Mission: title: str description: str rules: str active: bool = True class CampaignIC(gl.Contract): title: str goal: str knowledge_base: str rules: str style: str creator_address: Address campaign_duration_periods: u256 period_length_days: u256 missions: TreeMap[str, Mission] start_date: str end_date: str x_id_contract: str submissions: DynArray[Submission] tweet_store: VecDB[np.float32, typing.Literal[384], VectorStoreValue] token: str # Token address on the target chain bridge_sender: Address distribution_contract_chain_id: u256 distribution_contract_address: str processed_periods: TreeMap[u256, bool] minimum_followers: u256 maximum_followers: u256 whitelisted_submitters: DynArray[str] def __init__( self, title: str, goal: str, knowledge_base: str, rules: str, style: str, start_datetime_iso: str, campaign_duration_periods: int, period_length_days: int, missions: dict[str, dict[str, typing.Any]], token: str, x_id_contract: str, creator_address: Address, bridge_sender: Address, distribution_contract_chain_id: int, distribution_contract_address: str, minimum_followers: int, maximum_followers: int, whitelisted_submitters: str, ): self.title = title self.goal = goal self.knowledge_base = knowledge_base self.rules = rules self.style = style self.campaign_duration_periods = campaign_duration_periods self.period_length_days = period_length_days start_date = ( datetime.fromisoformat(start_datetime_iso) if start_datetime_iso else datetime.now(timezone.utc) ) self.start_date = start_date.isoformat() self.end_date = ( start_date + timedelta(days=campaign_duration_periods * period_length_days) ).isoformat() if missions: for mission_id, mission_data in missions.items(): self.missions[mission_id] = Mission(**mission_data) self.x_id_contract = x_id_contract self.creator_address = creator_address self.bridge_sender = bridge_sender self.distribution_contract_chain_id = distribution_contract_chain_id self.distribution_contract_address = distribution_contract_address self.token = token self.minimum_followers = minimum_followers self.maximum_followers = maximum_followers self.whitelisted_submitters = ( [username.strip() for username in whitelisted_submitters.split(",")] if whitelisted_submitters else [] ) def set_bridge_sender(self, bridge_sender: Address): self.require_creator() self.bridge_sender = bridge_sender @gl.public.write def set_distribution_contract(self, chain_id: int, address: str): """Updates the distribution contract chain ID and address. Can only be called by the creator.""" self.require_creator() self.distribution_contract_chain_id = chain_id self.distribution_contract_address = address def get_embedding_generator(self): return genlayermodelwrappers.SentenceTransformer("all-MiniLM-L6-v2") def get_embedding( self, txt: str ) -> np.ndarray[tuple[typing.Literal[384]], np.dtypes.Float32DType]: return self.get_embedding_generator()(txt) def _get_day_index(self, date: datetime) -> int: return (date - datetime.fromisoformat(self.start_date)).days def _get_period_index(self, date: datetime) -> int: total_days = self._get_day_index(date) # integer division by self.period_length_days return total_days // self.period_length_days @gl.public.view def get_now_period_index(self) -> int: return self._get_period_index(datetime.now(timezone.utc)) def _require_now_in_valid_range(self): now_period_index = self.get_now_period_index() self._require_in_valid_range(now_period_index) def _require_in_valid_range(self, period_index: int): if period_index < 0: raise ValueError( f"Date {datetime.now(timezone.utc)} is before the campaign start date {self.start_date} by {period_index} periods." ) if period_index >= self.campaign_duration_periods: raise ValueError("Campaign has already ended.") def _require_now_before_end_date(self): now_period_index = self.get_now_period_index() if now_period_index >= self.campaign_duration_periods: raise ValueError("Campaign has already ended.") def require_creator(self): if gl.message.sender_address != self.creator_address: raise ValueError("Only the creator can perform this action.") # CRUD operations @gl.public.write def submit_mission( self, mission_id: str, mission_title: str, mission_description: str, mission_rules: str, ): self.require_creator() self._require_now_before_end_date() if mission_id in self.missions: raise ValueError("Mission id already exists.") # Create and store the mission self.missions[mission_id] = Mission( title=mission_title, description=mission_description, rules=mission_rules, active=True, ) print(f"Mission {mission_id} submitted: {self.missions[mission_id]}") @gl.public.write def deactivate_mission(self, mission_uuid: str): self.require_creator() if mission_uuid not in self.missions: raise ValueError(f"Mission {mission_uuid} not found") mission = self.missions[mission_uuid] mission.active = False print(f"Mission {mission_uuid} deactivated") @gl.public.write def submit_tweet(self, mission_id: str, tweet_url_or_id: str): now = datetime.now(timezone.utc) tweet_id = extract_tweet_id(tweet_url_or_id) now_period_index = self._get_period_index(now) # Validations self._require_now_in_valid_range() if mission_id not in self.missions: raise ValueError(f"Mission {mission_id} not found") mission = self.missions[mission_id] if not mission.active: raise ValueError(f"Mission {mission_id} is not active.") tweet_data = get_tweet_data(tweet_id) user = get_tweet_user_handle(tweet_data) if ( len(self.whitelisted_submitters) > 0 and user not in self.whitelisted_submitters ): raise ValueError( f"User {user} is not whitelisted to submit to this campaign." ) user_followers = get_followers_of(user_usernames=[user]) if user_followers < self.minimum_followers: raise ValueError( f"User {user} has less than {self.minimum_followers} followers." ) if user_followers > self.maximum_followers: raise ValueError( f"User {user} has more than {self.maximum_followers} followers." ) # Check for duplicates # TODO: dicts might be faster if any( submission.mission_id == mission_id and submission.tweet_id == tweet_id for submission in self.submissions ): raise ValueError(f"Tweet {tweet_id} already submitted.") if any( submission.mission_id == mission_id and submission.user == user and submission.period_index == now_period_index for submission in self.submissions ): raise ValueError( f"Only one submission per period is allowed for user {user} on mission {mission_id}" ) # Fetch from X tweet_author_id_to_reply_text = fetch_tweet_replies(tweet_id) print(f"Fetched tweet replies: {tweet_author_id_to_reply_text}") tweet_content, _ = get_tweet_content(tweet_data) print(f"Fetched tweet content: {tweet_content}") tweet_metrics = fetch_tweet_metrics(tweet_data) print(f"Fetched tweet metrics: {tweet_metrics}") tweet_followers_of_repliers = get_followers_of( user_ids=tweet_author_id_to_reply_text.keys() ) print(f"Fetched tweet followers repliers: {tweet_followers_of_repliers}") self.submit_tweet_llm_analysis( user, tweet_id, now_period_index, mission_id, tweet_content, tweet_metrics, tweet_author_id_to_reply_text.values(), tweet_followers_of_repliers, ) @gl.public.write # TODO: hide this method from the outside world on production def submit_tweet_llm_analysis( self, user: str, tweet_id: str, now_period_index: int, mission_id: str, tweet_content: str, tweet_metrics: dict[str, int], tweet_replies: list[str], followers_of_repliers: int, ): """ This method is used for decoupling the LLM analysis from the tweet retrieval """ # LLM analysis atemporal_points, atemporal_points_analysis = ( self.get_atemporal_points_for_tweet(mission_id, tweet_content) ) print(f"Atemporal points: {atemporal_points}") print(f"Atemporal points analysis: {atemporal_points_analysis}") temporal_points, temporal_points_analysis = self.get_temporal_points_for_tweet( tweet_content, tweet_metrics, tweet_replies, followers_of_repliers, ) if atemporal_points == 0: temporal_points = 0 print(f"Temporal points: {temporal_points}") print(f"Temporal points analysis: {temporal_points_analysis}") # Store submission data new_submission = Submission( mission_id=mission_id, user=user, tweet_id=tweet_id, period_index=now_period_index, timestamp=datetime.now(timezone.utc).isoformat(), atto_atemporal_points=atemporal_points, atto_temporal_points=temporal_points, analysis=temporal_points_analysis + atemporal_points_analysis, ) print(f"New submission: {new_submission}") self.submissions.append(new_submission) emb = self.get_embedding(tweet_content) self.tweet_store.insert(emb, VectorStoreValue(text=tweet_content, id=tweet_id)) # Atemporal analysis @gl.public.write def get_atemporal_points_for_tweet(self, mission_id: str, tweet_content: str): content_alignment_analysis = self.get_content_alignment( mission_id, tweet_content ) print(f"Content alignment: {content_alignment_analysis}") engagement_potential_analysis = self.get_engagement_potential( mission_id, tweet_content ) print(f"Engagement potential: {engagement_potential_analysis}") information_accuracy_analysis = self.get_information_accuracy( mission_id, tweet_content ) print(f"Information accuracy: {information_accuracy_analysis}") campaign_compliance_analysis = self.get_campaign_compliance( mission_id, tweet_content ) print(f"Campaign compliance: {campaign_compliance_analysis}") originality_and_authenticity_analysis = self.get_originality_and_authenticity( mission_id, tweet_content ) print(f"Originality and authenticity: {originality_and_authenticity_analysis}") technical_quality_analysis = self.get_technical_quality( mission_id, tweet_content ) print(f"Technical quality: {technical_quality_analysis}") all_analyses = [ content_alignment_analysis, engagement_potential_analysis, information_accuracy_analysis, campaign_compliance_analysis, originality_and_authenticity_analysis, technical_quality_analysis, ] print("Checking minimum score") for analysis in all_analyses: if analysis.atto_score < MINIMUM_CATEGORY_SCORE_ATTO: print( f"Atemporal score below minimum: {analysis.atto_score} < {MINIMUM_CATEGORY_SCORE_ATTO}" ) return (0, all_analyses) total_score = sum( # TODO: improve formula, probably using weights [analysis.atto_score for analysis in all_analyses] ) print(f"Total atemporal score: {from_atto(total_score)}") return (total_score, all_analyses) def get_content_alignment(self, mission_id: str, tweet_content: str): mission = self.missions[mission_id] task = f""" You are an expert content analyst specializing in campaign messaging alignment. Your task is to analyze a tweet's alignment with campaign values and provide a comprehensive score from 0 to 10. Scoring criteria: 1. Message Accuracy - How accurately does the tweet represent the campaign's core message? - Are key features and benefits correctly portrayed? 2. Terminology Usage - Is campaign-specific terminology used correctly? - Are technical terms accurately represented? 3. Value Proposition - How well are the core value propositions integrated? - Is the campaign's unique selling point clearly conveyed? 4. Brand Consistency - Does the messaging align with official campaign style? - Is the tone consistent with campaign guidelines? 5. Target Audience Alignment - Does the content resonate with the intended audience? - Are audience-specific pain points addressed? Here's the input below: {tweet_content} {self.knowledge_base} {self.goal} {mission.title} {mission.description} {mission.rules} Output format - JSON: {{ "analysis": , "score": }} It is mandatory that you respond only using the JSON format above, nothing else. Don't include any other words or characters, your output must be only JSON without any formatting prefix or suffix. This result should be perfectly parsable by a JSON parser without errors. """ def leader_fn(): result = gl.exec_prompt(task) print(f"Content alignment result: {result}") # Parse the JSON result parsed_result = _decode_json_resilient(result) return parsed_result def validator_fn( leaders: advanced.ContractReturn | Rollback | advanced.ContractError, ) -> bool: my_res, leaders_res = ( advanced.validator_handle_rollbacks_and_errors_default( leader_fn, leaders ) ) return abs(my_res["score"] - leaders_res["score"]) <= MAX_SCORE_DIFFERENCE analysis = advanced.run_nondet(leader_fn, validator_fn).get() analysis_record = AnalysisRecord( atto_score=to_atto(analysis["score"]), analysis=analysis["analysis"], category="Content Alignment", atto_max_score=to_atto(10), ) return analysis_record def get_engagement_potential(self, mission_id: str, tweet_content: str): mission = self.missions[mission_id] task = f""" You are an expert social media strategist specializing in engagement analysis. Your task is to evaluate a tweet's potential to drive audience engagement and provide a comprehensive score from 0 to 10. Scoring criteria: 1. Hook Effectiveness - Does the opening grab attention? - Is there a compelling reason to engage? 2. Call-to-Action Quality - Is there a clear, actionable CTA? - How compelling is the invitation to engage? 3. Content Structure - Is the content easily digestible? - Are formatting elements used effectively? 4. Conversation Potential - Does the content invite discussion? - Are there elements that encourage sharing? 5. Value Delivery - Does the tweet provide clear value to readers? - Is there a reason for followers to save/bookmark? Here's the input below: {tweet_content} {mission.title} {mission.description} {mission.rules} Output format - JSON: {{ "analysis": , "score": }} It is mandatory that you respond only using the JSON format above, nothing else. Don't include any other words or characters, your output must be only JSON without any formatting prefix or suffix. This result should be perfectly parsable by a JSON parser without errors. """ def leader_fn(): result = gl.exec_prompt(task) print(f"Engagement potential result: {result}") # Parse the JSON result parsed_result = _decode_json_resilient(result) return parsed_result def validator_fn( leaders: advanced.ContractReturn | Rollback | advanced.ContractError, ) -> bool: my_res, leaders_res = ( advanced.validator_handle_rollbacks_and_errors_default( leader_fn, leaders ) ) return abs(my_res["score"] - leaders_res["score"]) <= MAX_SCORE_DIFFERENCE analysis = advanced.run_nondet(leader_fn, validator_fn).get() analysis_record = AnalysisRecord( atto_score=to_atto(analysis["score"]), analysis=analysis["analysis"], category="Engagement Potential", atto_max_score=to_atto(10), ) return analysis_record def get_information_accuracy(self, mission_id: str, tweet_content: str): mission = self.missions[mission_id] task = f""" You are an expert fact-checker specializing in campaign information verification. Your task is to assess the factual accuracy of this tweet and provide a comprehensive score from 0 to 10. Scoring criteria: 1. Technical Accuracy - Are technical terms used correctly? - Are features and capabilities accurately described? 2. Documentation Alignment - Does the content match official campaign documentation? - Are claims consistent with campaign materials? 3. Data & Statistics - Are numbers and statistics accurately represented? - Are quantitative claims verifiable? 4. Context Accuracy - Is information presented in proper context? - Are relationships between concepts correctly portrayed? 5. Claims Verification - Are statements supported by campaign materials? - Is there absence of misleading or false information? Here's the input below: {tweet_content} {self.knowledge_base} {mission.title} {mission.description} {mission.rules} Output format - JSON: {{ "analysis": , "score": }} """ def leader_fn(): result = gl.exec_prompt(task) print(f"Information accuracy result: {result}") # Parse the JSON result parsed_result = _decode_json_resilient(result) return parsed_result def validator_fn( leaders: advanced.ContractReturn | Rollback | advanced.ContractError, ) -> bool: my_res, leaders_res = ( advanced.validator_handle_rollbacks_and_errors_default( leader_fn, leaders ) ) return abs(my_res["score"] - leaders_res["score"]) <= MAX_SCORE_DIFFERENCE analysis = advanced.run_nondet(leader_fn, validator_fn).get() analysis_record = AnalysisRecord( atto_score=to_atto(analysis["score"]), analysis=analysis["analysis"], category="Information Accuracy", atto_max_score=to_atto(10), ) return analysis_record def get_campaign_compliance(self, mission_id: str, tweet_content: str): mission = self.missions[mission_id] task = f""" You are an expert compliance analyst specializing in campaign guidelines. Your task is to evaluate this tweet's adherence to campaign requirements and provide a comprehensive score from 0 to 10. Scoring criteria: 1. Required Elements - Are all mandatory hashtags present? - Are required mentions included? 2. Format Compliance - Does the tweet follow character limit guidelines? - Are media elements properly formatted? 3. Guideline Adherence - Does the content follow campaign style guidelines? - Are brand voice requirements met? 4. Disclosure Compliance - Are necessary disclaimers present? - Are sponsorship/partnership declarations included if required? 5. Content Restrictions - Does the tweet avoid prohibited content? - Are sensitive topics handled appropriately? Here's the input below: {tweet_content} {self.rules} {mission.title} {mission.description} {mission.rules} Output format - JSON: {{ "analysis": , "score": }} It is mandatory that you respond only using the JSON format above, nothing else. Don't include any other words or characters, your output must be only JSON without any formatting prefix or suffix. This result should be perfectly parsable by a JSON parser without errors. """ def leader_fn(): result = gl.exec_prompt(task) print(f"Campaign compliance result: {result}") # Parse the JSON result parsed_result = _decode_json_resilient(result) return parsed_result def validator_fn( leaders: advanced.ContractReturn | Rollback | advanced.ContractError, ) -> bool: my_res, leaders_res = ( advanced.validator_handle_rollbacks_and_errors_default( leader_fn, leaders ) ) return abs(my_res["score"] - leaders_res["score"]) <= MAX_SCORE_DIFFERENCE analysis = advanced.run_nondet(leader_fn, validator_fn).get() analysis_record = AnalysisRecord( atto_score=to_atto(analysis["score"]), analysis=analysis["analysis"], category="Campaign Compliance", atto_max_score=to_atto(10), ) return analysis_record def get_originality_and_authenticity(self, mission_id: str, tweet_content: str): # Get similar tweets embedding = self.get_embedding(tweet_content) similar_tweets_vectors = list(self.tweet_store.knn(embedding, 5)) similar_tweets = [vector.value.text for vector in similar_tweets_vectors] # Early return with perfect score if no similar tweets found if not similar_tweets: return AnalysisRecord( atto_score=to_atto(10), # Maximum score since no duplicates analysis="No similar tweets found - content is completely original", category="Originality and Authenticity", atto_max_score=to_atto(10), ) mission = self.missions[mission_id] task = f""" You are an expert content authenticity analyst specializing in originality assessment. Your task is to evaluate this tweet's uniqueness and authentic voice while comparing it to similar content, providing a comprehensive score from 0 to 10. Scoring criteria: 1. Unique Perspective - Does the tweet offer a fresh angle or viewpoint? - Is the approach different from standard campaign messaging? 2. Personal Voice - Is there authentic personal experience or insight? - Does the voice feel genuine rather than templated? 3. Content Distinctiveness - How different is it from similar tweets provided? - Does it avoid common phrases and clichés? 4. Language Authenticity - Is the language natural and conversational? - Does it avoid overly promotional or marketing speak? 5. Creative Expression - Are campaign elements presented in an innovative way? - Does the content show original thinking? Here's the input below: {tweet_content} {similar_tweets} {mission.title} {mission.description} {mission.rules} Output format - JSON: {{ "analysis": , "score": }} It is mandatory that you respond only using the JSON format above, nothing else. Don't include any other words or characters, your output must be only JSON without any formatting prefix or suffix. This result should be perfectly parsable by a JSON parser without errors. """ def leader_fn(): result = gl.exec_prompt(task) print(f"Originality and authenticity result: {result}") # Parse the JSON result parsed_result = _decode_json_resilient(result) return parsed_result def validator_fn( leaders: advanced.ContractReturn | Rollback | advanced.ContractError, ) -> bool: my_res, leaders_res = ( advanced.validator_handle_rollbacks_and_errors_default( leader_fn, leaders ) ) return abs(my_res["score"] - leaders_res["score"]) <= MAX_SCORE_DIFFERENCE analysis = advanced.run_nondet(leader_fn, validator_fn).get() analysis_record = AnalysisRecord( atto_score=to_atto(analysis["score"]), analysis=analysis["analysis"], category="Originality and Authenticity", atto_max_score=to_atto(10), ) return analysis_record def get_technical_quality(self, mission_id: str, tweet_content: str): mission = self.missions[mission_id] task = f""" You are an expert content quality analyst specializing in technical execution. Your task is to evaluate the technical proficiency of this tweet's composition and provide a comprehensive score from 0 to 10. Scoring criteria: 1. Language Mechanics - Is the grammar correct and appropriate? - Are spelling and punctuation error-free? 2. Structure & Formatting - Is the content well-organized and readable? - Are line breaks, spacing, and formatting used effectively? 3. Platform Optimization - Are Twitter features used appropriately? - Is the content optimized for the platform's constraints? 4. Media Integration - Are links, images, or other media properly formatted? - Do media elements enhance rather than detract? 5. Thread Composition - If part of a thread, is it properly structured? - Are connections between tweets clear and logical? Here's the input below: {tweet_content} {mission.title} {mission.description} {mission.rules} Output format - JSON: {{ "analysis": , "score": }} It is mandatory that you respond only using the JSON format above, nothing else. Don't include any other words or characters, your output must be only JSON without any formatting prefix or suffix. This result should be perfectly parsable by a JSON parser without errors. """ def leader_fn(): result = gl.exec_prompt(task) print(f"Technical quality result: {result}") # Parse the JSON result parsed_result = _decode_json_resilient(result) return parsed_result def validator_fn( leaders: advanced.ContractReturn | Rollback | advanced.ContractError, ) -> bool: my_res, leaders_res = ( advanced.validator_handle_rollbacks_and_errors_default( leader_fn, leaders ) ) return abs(my_res["score"] - leaders_res["score"]) <= MAX_SCORE_DIFFERENCE analysis = advanced.run_nondet(leader_fn, validator_fn).get() analysis_record = AnalysisRecord( atto_score=to_atto(analysis["score"]), analysis=analysis["analysis"], category="Technical Quality", atto_max_score=to_atto(10), ) return analysis_record # Temporal analysis @gl.public.write def get_temporal_points_for_tweet( self, tweet_content: str, tweet_metrics: dict[str, int], tweet_replies: list[str], followers_of_repliers: int, ): """ The sum of: - log base 10 of the sum of the tweet's likes and bookmarks - quality of replies (LLM analysis from 0-10) (TODO: should we analyze quotes too?) - log base 10 of the followers of the users who replied (TODO: how do we mix in the verified factor?) """ likes_and_bookmarks = ( tweet_metrics["like_count"] + tweet_metrics["bookmark_count"] ) log_likes_and_bookmarks = ( log10(likes_and_bookmarks) if likes_and_bookmarks > 0 else 0 ) print(f"Log of likes and bookmarks: {log_likes_and_bookmarks}") quality_of_replies_analysis = self.get_quality_of_replies( tweet_content, tweet_replies ) quality_of_replies = from_atto(quality_of_replies_analysis.atto_score) print(f"Quality of replies: {quality_of_replies}") log_followers_of_repliers = ( log10(followers_of_repliers) if followers_of_repliers > 0 else 0 ) print(f"Log of followers of repliers: {log_followers_of_repliers}") total_temporal_points = ( log_likes_and_bookmarks + quality_of_replies + log_followers_of_repliers ) print(f"Total temporal points: {total_temporal_points}") return ( to_atto(total_temporal_points), [ AnalysisRecord( category="Likes and Bookmarks", atto_score=to_atto(log_likes_and_bookmarks), analysis=f"Log of likes and bookmarks ({likes_and_bookmarks}): {log_likes_and_bookmarks}", atto_max_score=0, ), quality_of_replies_analysis, AnalysisRecord( category="Followers of Repliers", atto_score=to_atto(log_followers_of_repliers), analysis=f"Log of followers of repliers ({followers_of_repliers}): {log_followers_of_repliers}", atto_max_score=0, ), ], ) def get_quality_of_replies(self, tweet_content: str, tweet_replies: list[str]): """ Analyzes the quality of replies to a tweet. """ if not tweet_replies: return AnalysisRecord( atto_score=0, analysis="No replies to analyze", category="Reply Quality", atto_max_score=to_atto(10), ) task = f""" You are an expert social media analyst specializing in evaluating the quality of Twitter/X replies. Your task is to analyze replies to a tweet and provide a comprehensive quality score from 0 to 10. Scoring criteria: 1. Relevance - How well do the replies address the original tweet's content? - Are the responses on-topic and contextually appropriate? 2. Constructiveness - Do the replies contribute meaningfully to the discussion? - Are there thoughtful counterpoints or additional perspectives? 3. Civility - Is the tone respectful and professional? - Are the replies free from harassment, hate speech, or excessive negativity? 4. Engagement Quality - Is there meaningful interaction between users? - Are there substantive follow-up discussions? 5. Information Value - Do the replies contain accurate, verifiable information? - Are claims supported by evidence when appropriate? Here's the input below: {tweet_content} {tweet_replies} {self.knowledge_base} Output format - JSON: {{ "analysis": , "score": }} It is mandatory that you respond only using the JSON format above, nothing else. Don't include any other words or characters, your output must be only JSON without any formatting prefix or suffix. This result should be perfectly parsable by a JSON parser without errors. """ def leader_fn(): result = gl.exec_prompt(task) print(f"Quality of replies result: {result}") # Parse the JSON result parsed_result = _decode_json_resilient(result) return parsed_result def validator_fn( leaders: advanced.ContractReturn | Rollback | advanced.ContractError, ) -> bool: my_res, leaders_res = ( advanced.validator_handle_rollbacks_and_errors_default( leader_fn, leaders ) ) return abs(my_res["score"] - leaders_res["score"]) <= MAX_SCORE_DIFFERENCE analysis = advanced.run_nondet(leader_fn, validator_fn).get() analysis_record = AnalysisRecord( atto_score=to_atto(analysis["score"]), analysis=analysis["analysis"], category="Reply Quality", atto_max_score=to_atto(10), ) return analysis_record # Points distribution @gl.public.view def get_scoreboard(self) -> dict[str, dict[str, str]]: """ Returns a dictionary with the period index as the key and a dictionary with the user and their total points for that period as the value. The key and values must be `str` due to calldata restrictions. """ scoreboard = self._get_scoreboard() return { str(period_index): { user: str(points) for user, points in period_scores.items() } for period_index, period_scores in scoreboard.items() } def _get_scoreboard(self) -> dict[int, dict[str, int]]: """ Returns a dictionary with the period index as the key and a dictionary with the user and their total points for that period as the value. """ scoreboard = { period_index: { user: sum( submission.atto_atemporal_points + submission.atto_temporal_points for submission in user_submissions ) for user, user_submissions in groupby( period_submissions, lambda s: s.user ) } for period_index, period_submissions in groupby( self.submissions, key=lambda s: s.period_index ) } print(f"Scoreboard: {scoreboard}") return scoreboard # TODO: review if rounding causes trouble. Maybe we should do the rounding in Base? # Edgars proposal: multiply by a big number to be able to have better precision, and return a u256 @gl.public.view def get_period_distribution(self, period_index: int) -> dict[str, int]: """ Distribution for a given period. The sum of all values is 1e18. The distribution for each user is the sum of the points they have for that period divided by the total points for that period. """ if period_index < 0 or period_index >= self.campaign_duration_periods: raise ValueError("Invalid day index.") scoreboard = self._get_scoreboard() total_points_for_period = sum(scoreboard[period_index].values()) period_distribution = { user: int(points / total_points_for_period) for user, points in scoreboard[ period_index ].items() # Proably can be optimized if we only get the scoreboard for this date } print(f"Period distribution for period {period_index}: {period_distribution}") return period_distribution @gl.public.view def get_campaign_info(self): campaign_info = { "title": self.title, "goal": self.goal, "creator_address": self.creator_address.as_hex, # genlayer-js 0.4.8 doesn't handle the GenVM Address type correctly "style": self.style, "rules": self.rules, "duration_periods": self.campaign_duration_periods, "period_length_days": self.period_length_days, "start_date": self.start_date, "end_date": self.end_date, "missions": self.missions, "submissions": self.submissions, "knowledge_base": self.knowledge_base, "x_id_contract": self.x_id_contract, "bridge_sender": self.bridge_sender.as_hex, "distribution_contract_chain_id": self.distribution_contract_chain_id, "distribution_contract_address": self.distribution_contract_address, "token": self.token, "minimum_followers": self.minimum_followers, "maximum_followers": self.maximum_followers, "whitelisted_submitters": self.whitelisted_submitters, } print(f"Campaign info: {campaign_info}") return campaign_info @gl.public.view def get_submissions(self, mission_id: str) -> list[typing.Any]: """ TODO: change return type to SubmissionView after new GenVM feature of allowing dataclass serialization """ return [ submission for submission in self.submissions if submission.mission_id == mission_id ] @gl.public.write def resubmit_tweet(self, mission_id: str, tweet_url_or_id: str): now_period_index = self.get_now_period_index() self.resubmit_tweet_inner(mission_id, tweet_url_or_id, now_period_index) def resubmit_tweet_inner( self, mission_id: str, tweet_url_or_id: str, now_period_index: int ): """ TODO: remove from public.write once we finish testing TODO: should we store the tweet content in the submission to not fetch it from X again? """ # Validations self._require_in_valid_range(now_period_index) if mission_id not in self.missions: raise ValueError(f"Mission {mission_id} not found") mission = self.missions[mission_id] if not mission.active: raise ValueError(f"Mission {mission_id} is not active.") tweet_id = extract_tweet_id(tweet_url_or_id) tweet_data = get_tweet_data(tweet_id) user = get_tweet_user_handle(tweet_data) # Check for existing submission submissions_sorted = sorted( ( submission for submission in self.submissions if submission.mission_id == mission_id and submission.tweet_id == tweet_id and submission.user == user ), key=lambda x: x.period_index, ) if not submissions_sorted: raise ValueError( f"No existing submission found for resubmission for mission {mission_id} and tweet {tweet_id}" ) last_submission = submissions_sorted[-1] # Check for duplicates if last_submission.period_index >= now_period_index: raise ValueError( f"Only one resubmission per period is allowed for mission {mission_id} and tweet {tweet_id}" ) # Fetch from X tweet_metrics = fetch_tweet_metrics(tweet_data) print(f"Fetched tweet metrics: {tweet_metrics}") tweet_author_id_to_reply_text = fetch_tweet_replies(tweet_id) # Calculate temporal points delta new_temporal_points, temporal_points_explanation = ( self.get_temporal_points_for_tweet( tweet_id, tweet_metrics, tweet_author_id_to_reply_text.values(), get_followers_of(user_ids=tweet_author_id_to_reply_text.keys()), ) ) accumulated_temporal_points = sum( s.atto_temporal_points for s in submissions_sorted ) temporal_points_delta = max( new_temporal_points - accumulated_temporal_points, 0 ) # Store resubmission data new_submission = Submission( mission_id=mission_id, user=user, tweet_id=tweet_id, period_index=now_period_index, timestamp=datetime.now(timezone.utc).isoformat(), atto_atemporal_points=0, # Resubmissions have 0 atemporal points atto_temporal_points=temporal_points_delta, analysis=temporal_points_explanation, ) print(f"New resubmission: {new_submission}") self.submissions.append(new_submission) @gl.public.write def bridge_period_distribution(self, period_index: int): """Bridge a period's distribution to the EVM campaign contract""" # Verify the period is complete if period_index <= self.get_now_period_index(): raise ValueError("Period is not complete") if ( period_index in self.processed_periods and self.processed_periods[period_index] ): raise ValueError("Period already processed") # Get the distribution data distribution = self.get_period_distribution(period_index) # Encode the message data abi = [u32, Address, Address, bytes] encoder = genvm_eth.MethodEncoder("", abi, bool) # Prepare the message data message_data = [ distribution.keys(), distribution.values(), period_index, ] message_bytes = encoder.encode_call(message_data) # TODO: currently need to remove the first 4 bytes of the message message_bytes = message_bytes[4:] # Send to bridge bridge_contract = gl.ContractAt(self.bridge_sender) bridge_contract.emit().send_message( self.distribution_contract_chain_id, self.distribution_contract_address, message_bytes, ) # Mark period as processed self.processed_periods[period_index] = True # X API def extract_tweet_id(tweet_url_or_id: str) -> str: tweet_url_or_id = tweet_url_or_id.strip() if re.match(r"^\d+$", tweet_url_or_id): return tweet_url_or_id else: # If it's not a number, we assume it's a URL return extract_tweet_id_from_url(tweet_url_or_id) def extract_tweet_id_from_url(tweet_url: str) -> str: # Extract tweet ID using regex match = re.search(r"status/(\d+)", tweet_url) if match: tweet_id = match.group(1) print(f"Extracted tweet ID: {tweet_id}") return tweet_id else: raise ValueError(f"Invalid Tweet URL: {tweet_url}") def get_tweet_data(tweet_id: str) -> dict: """ Get the tweet data in 1 call using the /tweet/{tweet_id} API NOTE: /tweet/{tweet_id} has low rate limit """ tweet_data = request_to_x( f"tweets/{tweet_id}", { "tweet.fields": "author_id,created_at,text,public_metrics", "expansions": "author_id", "user.fields": "username", }, ) return tweet_data def get_tweet_user_handle(tweet_data: dict) -> str: """ Gets the user handle for a tweet by querying the X API. """ # Get the author ID from the tweet data author_id = tweet_data["data"]["author_id"] tweet_id = tweet_data["data"]["id"] # Find the matching user object in the includes section for user in tweet_data["includes"]["users"]: if user["id"] == author_id: print(f"Found user: {user} for tweet {tweet_id}") return user["username"] raise ValueError(f"Could not find user data for tweet {tweet_id}") def get_tweet_content(tweet_data: dict) -> tuple[str, datetime]: tweet_content = tweet_data["data"]["text"] tweet_datetime = datetime.fromisoformat(tweet_data["data"]["created_at"]) return tweet_content, tweet_datetime def fetch_tweet_metrics(tweet_data: dict) -> dict[str, int]: return tweet_data["data"]["public_metrics"] def fetch_tweet_replies(tweet_id: str, max_results: int = 10) -> dict[str, str]: """ Returns a dictionary with the author ID as the key and the reply text as the value. TODO: this only returns the replies from the last 7 days. We should probably fetch all replies. """ params = { "query": f"conversation_id:{tweet_id}", "tweet.fields": "text", "max_results": max_results, "expansions": "author_id", "sort_order": "relevancy", } data = request_to_x("tweets/search/recent", params) print(data) if data["meta"]["result_count"] == 0: return {} return {reply["author_id"]: reply["text"] for reply in data["data"]} def get_followers_of(user_ids: list[str] = [], user_usernames: list[str] = []) -> int: """ Returns the sum of the number of followers of all users. """ if not user_ids and not user_usernames: return 0 user_params = {"user.fields": "public_metrics"} if len(user_ids) > 0: user_params["ids"] = ",".join(user_ids) endpoint = "users" else: user_params["usernames"] = ",".join(user_usernames) endpoint = "users/by" # Fetch user data including follower counts user_data = request_to_x(endpoint, user_params) print(f"{user_data=}") return sum(user["public_metrics"]["followers_count"] for user in user_data["data"]) # TODO: we should use equivalence principle for this once we have the X API proxy def request_to_x( endpoint: str, params: dict[typing.Any, typing.Any] ) -> dict[str, typing.Any]: proxy_url = "https://d-kol.vercel.app/api/twitter" base_url = f"{proxy_url}/{endpoint}" url = f"{base_url}?{urllib.parse.urlencode(params)}" def call_x_api() -> dict[str, typing.Any]: print(f"Requesting {url}") web_data = gl.get_webpage(url, mode="text") print(f"Response: '{web_data}'") # TODO: improve this to handle case when response is not a json # TODO: improve proxy server to return json in failure cases return json.loads(web_data) return gl.eq_principle_strict_eq(call_x_api) # Utils def _decode_json_resilient(s: str) -> dict: """ Decode a JSON string in a resilient manner. Args: s (str): The JSON string to decode. Returns: dict: The decoded JSON object. """ # Clean the string and replace boolean literals # Clean the string using translation table control_chars = "".join( [chr(x) for x in range(0, 32)] ) # All control characters (0-31) translation_table = str.maketrans("", "", control_chars) clean = ( _extract_json_from_string(s) .replace("True", "true") .replace("False", "false") .translate(translation_table) ) return json.loads(clean) def _extract_json_from_string(s: str) -> str: """ Extract a JSON object from a string. Args: s (str): The string potentially containing a JSON object. Returns: str: The extracted JSON string, or an empty string if no valid JSON is found. """ start_index = s.find("{") end_index = s.rfind("}") if start_index != -1 and end_index != -1 and start_index < end_index: return s[start_index : end_index + 1] else: return ""