RetroFed Forum
Your home away from modern life.
Code Highlighting Test
Post Reply
AuthorMessage
wjs018 piefed.wjs018.xyz wjs018@piefed.wjs018.xyz
Joined: 2025-07-15
Quote

This is a copy/paste of the code from this post to test the highlighting.

import os, sys  
import logging  

from app.plugins.hooks import hook  
from flask import current_app, g, Blueprint  
from app.utils import get_task_session  
from app.models import Site, Instance, Post, PostReply, User, ActivityPubLog, PostVote, PostReplyVote  
from sqlalchemy import text, func, literal_column, and_, or_  

from matplotlib.figure import Figure  
from matplotlib.colors import TABLEAU_COLORS  
import matplotlib.dates as mdates  
from datetime import datetime, timedelta  


PLUGIN_NAME = __name__.split('.')[-1]  
logger = logging.getLogger(__name__)  


def fetch_stats(session, obj, interval='day', starttime=None, endtime=None, count1=None, count2=None, sum2=None, createdcolumn=None):  
    if interval not in ['hour', 'day', 'week', 'month']:  
        interval = 'day'  
    if not starttime:  
        starttime = g.site.created_at  
    if not endtime:  
        endtime = datetime.now()  
    if not count1:  
        count1 = [obj.id != None]  
    if not createdcolumn:  
        createdcolumn = obj.created_at  

    timeseries = func.generate_series(starttime, endtime, f"1 {interval}").alias('timeseries')  
    one = literal_column("1")  

    if count2:  
        data = session.query(timeseries.column, func.count(one).filter(*count1), func.count(one).filter(*count2))  
    elif sum2:  
        data = session.query(timeseries.column, func.count(one).filter(*count1), func.sum(sum2))  
    else:  
        data = session.query(timeseries.column, func.count(one).filter(*count1))  
    data = data.select_from(timeseries).outerjoin(obj, func.date_trunc(interval, createdcolumn) == func.date_trunc(interval, timeseries.column))  
    data = data.group_by(timeseries.column).order_by(timeseries.column)  

    if interval == 'day':  
        data = data.all()[2:-2] # throw away first two days and today to remove outliers/unstable data  

    data2 = {'timestamp': [], 'count1': [], 'count2': [], 'zeros': []}  
    for row in data:  
        data2['timestamp'].append(row[0])  
        data2['count1'].append(row[1])  
        if count2:  
            data2['count2'].append(row[2])  
        data2['zeros'].append(0)  

    return data2  


@hook("cron_daily")  
def generate_charts_loop(data):  
    """Hook that runs on a periodic interval"""  
    from app import redis_client  
    with current_app.app_context(), get_task_session() as session:  
        g.site = session.query(Site).get(1)  

        languages = session.execute(text('SELECT language.name, COUNT(*) FROM post LEFT JOIN language ON post.language_id = language.id AND post.deleted = False GROUP BY name ORDER BY count DESC LIMIT 10')).all()  
        colors = list( TABLEAU_COLORS.keys() )  
        colors.insert(1, 'lightgray')  
        fig = Figure()  
        ax = fig.add_subplot()  
        labels = [r.name if r.name else 'Unknown' for r in languages]  
        hbars = ax.barh(labels, [r.count for r in languages], align='center', label=labels, color=colors)  
        ax.invert_yaxis()  
        ax.xaxis.set_visible(False)  
        ax.set_title('Top 10 post languages', y=0.0)  
        ax.text(0.99, -0.01, datetime.now().strftime('%Y-%m-%d %H:%M:%S') + '  on ' + g.site.name, verticalalignment='top', horizontalalignment='right', transform=ax.transAxes, color='darkgrey') #, fontsize=15  
        fig.tight_layout()  
        filename = "/var/lib/piefed/app/static/tmp/languages.svg"  
        fig.savefig(filename, format="svg")  

        software = session.execute(text('SELECT software, COUNT(*) FROM instance WHERE "gone_forever" IS False GROUP BY software ORDER BY count DESC LIMIT 15')).all()  
        myexplode = [0.2 if r.software == 'piefed' else 0 for r in software]  
        fig = Figure()  
        ax = fig.add_subplot()  
        colors = list( TABLEAU_COLORS.keys() )  
        hbars = ax.barh([r.software for r in software], [r.count for r in software], align='center', color=colors)  
        ax.bar_label(hbars, fmt='%d')  
        ax.invert_yaxis()  
        ax.set_title('Instances by Software', y=0.0)  
        ax.text(0.99, -0.01, datetime.now().strftime('%Y-%m-%d %H:%M:%S') + '  on ' + g.site.name, verticalalignment='top', horizontalalignment='right', transform=ax.transAxes, color='darkgrey') #, fontsize=15  
        fig.tight_layout()  
        fig.savefig("/var/lib/piefed/app/static/tmp/software.svg", format="svg") # , transparent=True  

        for software in ["mastodon", "lemmy", "peertube", "friendica", "piefed", "sharkey", "mbin"]:  
            version = session.execute(text('SELECT version, COUNT(*) FROM instance WHERE "software" = :software AND "gone_forever" IS False GROUP BY version ORDER BY count DESC'), {'software': software}).all()  
            fig = Figure()  
            ax = fig.add_subplot()  
            ax.pie([r.count for r in version], labels=[r.version for r in version])  
            ax.set_title(f"{software} versions", y=0.0)  
            ax.text(0.99, -0.01, datetime.now().strftime('%Y-%m-%d %H:%M:%S') + '  on ' + g.site.name, verticalalignment='top', horizontalalignment='right', transform=ax.transAxes, color='darkgrey') #, fontsize=15  
            fig.tight_layout()  
            fig.savefig(f"/var/lib/piefed/app/static/tmp/version_{software}.svg", format="svg")  

        for interval in ['day', 'week', 'month']: # hour  
            data = fetch_stats(session, Instance, interval)  
            data_gone = fetch_stats(session, Instance, interval, createdcolumn=Instance.last_seen, count1=[Instance.gone_forever == True])  
            fig = Figure()  
            ax = fig.subplots()  
            ax.fill_between(data['timestamp'], data['zeros'], data['count1'], step='pre', alpha=0.4)  
            ax.step(data['timestamp'], data['count1'], label="Instances discovered")  
            ax.fill_between(data_gone['timestamp'], data_gone['zeros'], [-x for x in data_gone['count1']], step='pre', alpha=0.4)  
            ax.step(data_gone['timestamp'], [-x for x in data_gone['count1']], label="Instances gone")  
            ax.plot([], [], ' ', label=f"(per {interval})")  
            ax.legend(loc="upper right")  
            fig.tight_layout()  
            fig.savefig(f"/var/lib/piefed/app/static/tmp/instances_{interval}.svg", format="svg")  

            data_posts = fetch_stats(session, Post, interval, count1=[and_(Post.deleted == False, Post.up_votes > 0)], count2=[Post.deleted == True])  
            data_replies = fetch_stats(session, PostReply, interval, count1=[and_(PostReply.deleted == False, PostReply.up_votes > 1)], count2=[PostReply.deleted == True])  
            posts_byvote = fetch_stats(session, Post, interval, count1=[and_(Post.deleted == False, Post.up_votes > 30)], count2=[and_(Post.deleted == False, Post.up_votes == 0)])  
            replies_byvote = fetch_stats(session, PostReply, interval, count1=[and_(PostReply.deleted == False, PostReply.up_votes > 10)], count2=[and_(PostReply.deleted == False, PostReply.up_votes <= 1)])  
            fig = Figure()  
            axs = fig.subplots(2)  
            axs[0].fill_between(data_posts['timestamp'], data_posts['zeros'], data_posts['count1'], step='pre', alpha=0.4)  
            axs[0].step(data_posts['timestamp'], data_posts['count1'], label = f"Posts (per {interval})")  
            axs[0].fill_between(data_posts['timestamp'], data_posts['zeros'], [-x for x in data_posts['count2']], step='pre', alpha=0.4)  
            axs[0].step(data_posts['timestamp'], [-x for x in data_posts['count2']], label="Posts removed")  
            axs[0].fill_between(posts_byvote['timestamp'], posts_byvote['zeros'], posts_byvote['count1'], step='pre', alpha=0.4)  
            axs[0].plot([], [], label=">30 upvotes")  
            axs[0].fill_between(posts_byvote['timestamp'], data_posts['count1'], [x + y for x, y in zip(data_posts['count1'], posts_byvote['count2'])], step='pre', alpha=0.4)  
            axs[0].plot([], [], label="0 upvotes")  
            axs[1].fill_between(data_replies['timestamp'], data_replies['zeros'], data_replies['count1'], step='pre', alpha=0.4)  
            axs[1].step(data_replies['timestamp'], data_replies['count1'], label = f"Comments (per {interval})")  
            axs[1].fill_between(data_replies['timestamp'], data_replies['zeros'], [-x for x in data_replies['count2']], step='pre', alpha=0.4)  
            axs[1].step(data_replies['timestamp'], [-x for x in data_replies['count2']], label= "Comments removed")  
            axs[1].fill_between(replies_byvote['timestamp'], replies_byvote['zeros'], replies_byvote['count1'], step='pre', alpha=0.4)  
            axs[1].plot([], [], label=">10 upvotes")  
            axs[1].fill_between(replies_byvote['timestamp'], data_replies['count1'], [x + y for x, y in zip(data_replies['count1'], replies_byvote['count2'])], step='pre', alpha=0.4)  
            axs[1].plot([], [], label="<=1 upvote")  
            axs[0].legend(loc="upper left")  
            axs[1].legend(loc="upper left")  
            fig.tight_layout()  
            fig.savefig(f"/var/lib/piefed/app/static/tmp/posts_{interval}.svg", format="svg")  

            data = fetch_stats(session, User, interval, createdcolumn=User.created, count1=[User.bot == False], count2=[and_(User.bot == False, User.deleted == False, User.banned == False, User.last_seen - User.created < timedelta(days=3), User.created < datetime.now() - timedelta(days=5))])  
            users_removed = fetch_stats(session, User, interval, createdcolumn=User.last_seen, count1=[or_(User.deleted == True, User.banned == True)])  
            fig = Figure()  
            ax = fig.subplots()  
            ax.fill_between(data['timestamp'], data['zeros'], data['count1'], step='pre', alpha=0.4)  
            ax.step(data['timestamp'], data['count1'], label="Users discovered")  
            ax.fill_between(users_removed['timestamp'], users_removed['zeros'], [-x for x in users_removed['count1']], step='pre', alpha=0.4)  
            ax.step(users_removed['timestamp'], [-x for x in users_removed['count1']], label="Users removed")  
            ax.fill_between(data['timestamp'], data['count1'], [x - y for x, y in zip(data['count1'], data['count2'])], step='pre', alpha=0.4)  
            ax.plot([], [], label="non-returning")  
            ax.plot([], [], ' ', label=f"(per {interval})")  
            # ToDo: show inactive  
            ax.legend(loc="upper right")  
            fig.tight_layout()  
            fig.savefig(f"/var/lib/piefed/app/static/tmp/users_{interval}.svg", format="svg")  

        starttime = datetime.now() - timedelta(days=3)  
        endtime = datetime.now()  
        interval = 'hour'  
        timeseries = func.generate_series(starttime, endtime, f"1 {interval}").alias('timeseries')  
        one = literal_column("1")  
        data = session.query(timeseries.column, func.count(one).filter(ActivityPubLog.id != None), ActivityPubLog.activity_type, ActivityPubLog.direction).select_from(timeseries).outerjoin(ActivityPubLog, func.date_trunc(interval, ActivityPubLog.created_at) == func.date_trunc(interval, timeseries.column)).group_by(timeseries.column, ActivityPubLog.activity_type, ActivityPubLog.direction).order_by(timeseries.column).all()[:-2]  
        timestamps = []  
        activity_types = []  
        values = dict()  
        values['in'] = dict()  
        values['out'] = dict()  
        for r in data:  
            timestamp_ = r[0]  
            value_ = r[1]  
            type_ = r[2]  
            direction_ = r[3]  
            if direction_ not in ['in', 'out']:  
                continue  
            if type_ not in values[direction_]:  
                values[direction_][type_] = dict()  
            values[direction_][type_][timestamp_] = value_  
            if timestamp_ not in timestamps:  
                timestamps.append(timestamp_)  
            if type_ not in activity_types:  
                activity_types.append(type_)  
        val2 = dict()  
        for direction in ['in', 'out']:  
            val2[direction] = dict()  
            for type_ in activity_types:  
                val2[direction][type_] = []  
                for i, ts in enumerate(timestamps):  
                    if (type_ in values[direction]) and (ts in values[direction][type_]):  
                        val2[direction][type_].append(values[direction][type_][ts])  
                    else:  
                        val2[direction][type_].append(0)  
        fig = Figure()  
        ax = fig.subplots()  
        defaultColors = [  
            "#3366CC", "#DC3912", "#FF9900", "#109618", "#990099", "#3B3EAC", "#0099C6",  
            "#DD4477", "#66AA00", "#B82E2E", "#316395", "#994499", "#22AA99", "#AAAA11",  
            "#6633CC", "#E67300", "#8B0707", "#329262", "#5574A6", "#651067"  
        ]  
        bottom = [0 for r in timestamps]  
        shownTypes = ['Announce', 'Create', 'Update', 'Delete', 'Undo Vote', 'Dislike', 'Like'] # , 'PeerTube View', 'Duplicate', 'User Ban', 'Add Mod/Sticky', 'Remove Mod/Sticky', 'Undo Delete', 'Post Lock', 'Accept', 'Follow', 'Flag', 'Unknown', 'Undo'  
        for type_, color in zip(shownTypes, defaultColors):  
            y2 = [x + y for x, y in zip(val2['in'][type_], bottom)]  
            ax.vlines(timestamps, bottom, y2, colors=color, linewidth=3, label=type_)  
            bottom = y2  
        ax.xaxis.set_major_formatter(mdates.DateFormatter('%H:%M'))  
        ax.legend(loc='center left')  
        fig.tight_layout()  
        fig.savefig(f"/var/lib/piefed/app/static/tmp/activity.svg", format="svg")  

        user = session.query(User).filter(User.id == 145).one()  
        interval = 'day'  
        starttime = max( datetime.now() - timedelta(days=180), user.created )  
        endtime = datetime.now()  
        data_posts = fetch_stats(session, Post, interval, starttime, endtime, count1=[Post.user_id == user.id])  # , sum2=Post.up_votes  
        data_replies = fetch_stats(session, PostReply, interval, starttime, endtime, count1=[PostReply.user_id == user.id])  
        data_post_votes = fetch_stats(session, PostVote, interval, starttime, endtime, count1=[and_(PostVote.user_id == user.id, PostVote.effect > 0)], count2=[and_(PostVote.user_id == user.id, PostVote.effect < 0)])  
        data_reply_votes = fetch_stats(session, PostReplyVote, interval, starttime, endtime, count1=[and_(PostReplyVote.user_id == user.id, PostReplyVote.effect > 0)], count2=[and_(PostReplyVote.user_id == user.id, PostReplyVote.effect < 0)])  
        data_votes = {'timestamp': [x for x, y in zip(data_post_votes['timestamp'], data_reply_votes['timestamp'])],  
                      'count1': [x + y for x, y in zip(data_post_votes['count1'], data_reply_votes['count1'])],  
                      'count2': [x + y for x, y in zip(data_post_votes['count2'], data_reply_votes['count2'])] }  
        data_post_votes_in = fetch_stats(session, PostVote, interval, starttime, endtime, count1=[and_(PostVote.author_id == user.id, PostVote.effect > 0)], count2=[and_(PostVote.author_id == user.id, PostVote.effect < 0)])  
        data_reply_votes_in = fetch_stats(session, PostReplyVote, interval, starttime, endtime, count1=[and_(PostReplyVote.author_id == user.id, PostReplyVote.effect > 0)], count2=[and_(PostReplyVote.author_id == user.id, PostReplyVote.effect < 0)])  
        data_votes_in = {'timestamp': [x for x, y in zip(data_post_votes_in['timestamp'], data_reply_votes_in['timestamp'])],  
                         'count1': [x + y for x, y in zip(data_post_votes_in['count1'], data_reply_votes_in['count1'])],  
                         'count2': [x + y for x, y in zip(data_post_votes_in['count2'], data_reply_votes_in['count2'])] }  
        fig = Figure()  
        axs = fig.subplots(3)  
        axs[0].set_title(f"{user.user_name}")  
        bottom = data_posts['zeros']  
        axs[0].vlines(data_posts['timestamp'], bottom, data_posts['count1'], colors="#3366CC", linewidth=2, label="Posts")  
        bottom = data_posts['count1']  
        axs[0].vlines(data_replies['timestamp'], bottom, [x+y for x, y in zip(data_posts['count1'], data_replies['count1'])], colors="#FF9900", linewidth=2, label="Comments")  
        axs[0].legend(loc="upper left")  
        bottom = [0 for r in data_votes['timestamp']]  
        axs[1].vlines(data_votes['timestamp'], bottom, data_votes['count1'], colors="#22AA99", linewidth=2, label="Upvotes")  
        axs[1].vlines(data_votes['timestamp'], bottom, [-x for x in data_votes['count2']], colors="#DD4477", linewidth=2, label="Downvotes")  
        axs[1].legend(loc="upper left")  
        bottom = [0 for r in data_votes_in['timestamp']]  
        axs[2].vlines(data_votes_in['timestamp'], bottom, data_votes_in['count1'], colors="#66AA00", linewidth=2, label="Upvotes")  
        axs[2].vlines(data_votes_in['timestamp'], bottom, [-x for x in data_votes_in['count2']], colors="#DC3912", linewidth=2, label="Downvotes")  
        fig.tight_layout()  
        fig.savefig(f"/var/lib/piefed/app/static/tmp/user_145.svg", format="svg")  


def plugin_info():  
    """Plugin metadata"""  
    return {  
        "name": "Matplotlib Charts",  
        "version": "0.0.1",  
        "description": "An experimental plugin that generates statistics charts",  
        "license": "AGPL-3.0",                                   # Must be AGPL-compatible  
        "source_url": "https://piefed.social/post/1308113",  # Required  
        "author": "hendrik"  
    }  
1   Original
Post Reply
Back to top