code:
import base64
import json
import os
from typing import Dict
from typing import Any
from typing import List
from dataclasses import dataclass
import flask
import requests
from datetime import datetime

POUSSETACHES_AUTH_KEY = os.getenv("POUSSETACHES_AUTH_KEY")

@dataclass
class Task:
    req_id: str
    tries: int

    payload: Any

@dataclass
class GetTask:
    payload: Any
    expected: int
    schedule: str
    task_id: str
    next_run: datetime
    tries: int
    url: str
    last_error_status_code: int
    last_error_body: str

class PousseTaches:
    def __init__(self, api_url: str, base_url: str) -> None:
        self.api_url = api_url
        self.base_url = base_url

    def push(
        self,
        payload: Any,
        path: str,
        expected: int = 200,
        schedule: str = "",
        delay: int = 0,
    ) -> str:
        # Encode our payload
        p = base64.b64encode(json.dumps(payload).encode()).decode()

        # Queue/push it
        resp = requests.post(
            self.api_url,
            json={
                "url": self.base_url + path,
                "payload": p,
                "expected": expected,
                "schedule": schedule,
                "delay": delay,
            },
        )
        resp.raise_for_status()

        return resp.headers["Poussetaches-Task-ID"]

    def parse(self, req: flask.Request) -> Task:
        if req.headers.get("Poussetaches-Auth-Key") != POUSSETACHES_AUTH_KEY:
            raise ValueError("Bad auth key")

        # Parse the "envelope"
        envelope = json.loads(req.data)
        print(req)
        print(f"envelope={envelope!r}")
        payload = json.loads(base64.b64decode(envelope["payload"]))

        return Task(
            req_id=envelope["req_id"], tries=envelope["tries"], payload=payload
        )  # type: ignore

    @staticmethod
    def _expand_task(t: Dict[str, Any]) -> None:
        try:
            t["payload"] = json.loads(base64.b64decode(t["payload"]))
        except json.JSONDecodeError:
            t["payload"] = base64.b64decode(t["payload"]).decode()

        if t["last_error_body"]:
            t["last_error_body"] = base64.b64decode(t["last_error_body"]).decode()

        t["next_run"] = datetime.fromtimestamp(float(t["next_run"] / 1e9))
        if t["last_run"]:
            t["last_run"] = datetime.fromtimestamp(float(t["last_run"] / 1e9))
        else:
            del t["last_run"]

    def _get(self, where: str) -> List[GetTask]:
        out = []

        resp = requests.get(self.api_url + f"/{where}")
        resp.raise_for_status()
        dat = resp.json()
        for t in dat["tasks"]:
            self._expand_task(t)
            out.append(
                GetTask(
                    task_id=t["id"],
                    payload=t["payload"],
                    expected=t["expected"],
                    schedule=t["schedule"],
                    tries=t["tries"],
                    url=t["url"],
                    last_error_status_code=t["last_error_status_code"],
                    last_error_body=t["last_error_body"],
                    next_run=t["next_run"],
                )
            )

        return out

    def get_cron(self) -> List[GetTask]:
        return self._get("cron")

    def get_success(self) -> List[GetTask]:
        return self._get("success")

    def get_waiting(self) -> List[GetTask]:
        return self._get("waiting")

    def get_dead(self) -> List[GetTask]:
        return self._get("dead")


code:
python-dateutil
libsass
tornado<6.0.0
gunicorn
piexif
requests
python-u2flib-server
Flask
Flask-WTF
pymongo
timeago
bleach
html2text
feedgen
itsdangerous
bcrypt
mf2py
passlib
git+https://github.com/erikriver/opengraph.git
git+https://github.com/tsileo/little-boxes.git
pyyaml
pillow
cachetools
python-dotenv

repo_name: nipos/nikisoftblog

license: mit


code:
[flake8]
max-line-length = 120


code:
cd $3
source env/bin/activate
FLASK_DEBUG=1 MICROBLOGPUB_DEBUG=1 FLASK_APP=app.py POUSSETACHES_AUTH_KEY=$1 flask run -p $2 --with-threads


code:
@font-face {
font-family: 'josefin-sans';
src:  url('josefin-sans.woff') format('woff'),
url('josefin-sans.svg#Josefin-Sans') format('svg'),
url('josefin-sans.eot'),
url('josefin-sans.eot?#iefix') format('embedded-opentype');
font-weight: normal;
font-style: normal;
}
@font-face {
font-family: 'josefin-sans';
src:  url('josefin-sans-bold.woff') format('woff'),
url('josefin-sans-bold.svg#Josefin-Sans-Bold') format('svg'),
url('josefin-sans-bold.eot'),
url('josefin-sans-bold.eot?#iefix') format('embedded-opentype');
font-weight: bold;
font-style: normal;
}


code:
* {
font-family:josefin-sans;
-webkit-box-sizing: border-box;
-moz-box-sizing: border-box;
box-sizing: border-box;
}
a {
text-decoration: none;
color: rgb(61, 146, 201);
}
a:hover,
a:focus {
text-decoration: underline;
}
h3 {
font-weight: 100;
}
/* LAYOUT CSS */
.pure-img-responsive {
max-width: 100%;
height: auto;
}
#layout {
padding: 0;
}
.header {
text-align: center;
top: auto;
margin: 3em auto;
}
.sidebar {
background: rgb(61, 79, 93);
color: #fff;
}
.brand-title,
.brand-tagline {
margin: 0;
}
.brand-title {
text-transform: uppercase;
}
.brand-tagline {
font-weight: 300;
color: rgb(176, 202, 219);
}
.nav-list {
margin: 0;
padding: 0;
list-style: none;
}
.nav-item {
display: inline-block;
*display: inline;
zoom: 1;
}
.nav-item a {
background: transparent;
border: 2px solid rgb(176, 202, 219);
color: #fff;
margin-top: 1em;
letter-spacing: 0.05em;
text-transform: uppercase;
font-size: 85%;
}
.nav-item a:hover,
.nav-item a:focus {
border: 2px solid rgb(61, 146, 201);
text-decoration: none;
}
.content-subhead {
text-transform: uppercase;
color: #aaa;
border-bottom: 1px solid #eee;
padding: 0.4em 0;
font-size: 80%;
font-weight: 500;
letter-spacing: 0.1em;
}
.content {
padding: 2em 1em 0;
}
.post {
padding-bottom: 2em;
}
.post-title {
font-size: 2em;
color: #222;
margin-bottom: 0.2em;
}
.post-avatar {
border-radius: 50px;
float: right;
margin-left: 1em;
}
.post-description {
font-family: Georgia, "Cambria", serif;
color: #444;
line-height: 1.8em;
}
.post-meta {
color: #999;
font-size: 90%;
margin: 0;
}
.post-category {
margin: 0 0.1em;
padding: 0.3em 1em;
color: #fff;
background: #999;
font-size: 80%;
}
.post-category-design {
background: #5aba59;
}
.post-category-pure {
background: #4d85d1;
}
.post-category-yui {
background: #8156a7;
}
.post-category-js {
background: #df2d4f;
}
.post-images {
margin: 1em 0;
}
.post-image-meta {
margin-top: -3.5em;
margin-left: 1em;
color: #fff;
text-shadow: 0 1px 1px #333;
}
.footer {
padding: 1em 0;
text-align:center;
}
.footer a {
color: #ccc;
}
.footer .pure-menu a:hover,
.footer .pure-menu a:focus {
background: none;
}
@media (min-width: 48em) {
.content {
padding: 2em 3em 0;
margin-left: 25%;
}
.header {
margin: 50% 2em 0;
text-align: right;
}
.sidebar {
position: fixed;
top: 0;
bottom: 0;
}
.footer {
position:absolute;
}
}
.pure-g {
letter-spacing:0;
}
.emoji {
width:20px;
margin-bottom:-5px;
}
.pure-button-group .action-form {
display:inline-block;
}
.pure-button-group .pure-button {
border-right:0;
}
.brand-title a {
color:#FFFFFF;
text-decoration:none;
}
a.pure-button,a.pure-menu-link {
text-decoration:none;
}
#admin-menu-wrapper {
position:fixed;
top:0;
right:0;
z-index:10;
width:auto;
background-color:#FFFFFF;
border-bottom-left-radius:5px;
}
.footer {
bottom:0;
width:100%;
}
.badge {
border-radius:5px;
padding:5px;
margin-right:-10px;
background-color:#b0cadb;
color:black;
}
.post-avatar,.actor-icon {
border-radius:5px;
}
aside {
background:#1f8dd6;
margin:auto;
margin-top:10%;
padding:1em;
border-radius:3px;
color:#fff;
font-size:25px;
}
aside input,aside textarea {
color:#000000;
font-size:20px;
}
.pure-button-success,
.pure-button-error,
.pure-button-warning,
.pure-button-secondary {
color: white;
border-radius: 4px;
text-shadow: 0 1px 1px rgba(0, 0, 0, 0.2);
}
.pure-button-success {
background: rgb(28, 184, 65);
}
.pure-button-error {
background: rgb(202, 60, 60);
}
.pure-button-warning {
background: rgb(223, 117, 20);
}
.pure-button-secondary {
background: rgb(66, 184, 221);
}
.img-attachment {
max-width:100%;
}
.actor-box {
border:2px solid #1f8dd6;
border-radius:10px;
padding:5px;
margin-bottom:10px;
width:400px;
line-height:initial;
min-height:64px;
}
.actor-button {
float:right;
}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block header %}
{% endblock %}
{% block content %}

{% include "header.html" %}

<data class="p-name" value="{{ config.USERNAME }}">

{{ text | safe }}

{{ me.summary }}

{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block title %}Admin - {{ config.NAME }}{% endblock %}
{% block content %}
{% include "header.html" %}

Database

Inbox size: {{ inbox_size }}

Outbox size: {{ outbox_size }}

Collections

followers: {{ col_followers }}

following: {{ col_following }}

liked: {{col_liked }}

{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block title %}Tasks - {{ config.NAME }}{% endblock %}
{% block content %}
{% include "header.html" %}

Cron

#
URL
Payload
Schedule
Next run
Response

{% for task in cron %}

{{ task.task_id }}
{{ task.url }} ({{ task.expected }})
{{ task.payload }}
{{ task.schedule }}
{{ task.next_run }}
Tries #{{ task.tries }}: {{ task.last_error_body }} ({{ task.last_error_status_code }})

{% endfor %}

Dead

#
URL
Payload
Next run
Response

{% for task in dead %}

{{ task.task_id }}
{{ task.url }} ({{ task.expected }})
{{ task.payload }}
{{ task.next_run }}
Tries #{{ task.tries }}: {{ task.last_error_body }} ({{ task.last_error_status_code }})

{% endfor %}

Waiting

#
URL
Payload
Next run
Response

{% for task in waiting %}

{{ task.task_id }}
{{ task.url }} ({{ task.expected }})
{{ task.payload }}
{{ task.next_run }}
Tries #{{ task.tries }}: {{ task.last_error_body }} ({{ task.last_error_status_code }})

{% endfor %}

Success

#
URL
Payload
Next run
Response

{% for task in success %}

{{ task.task_id }}
{{ task.url }} ({{ task.expected }})
{{ task.payload }}
{{ task.next_run }}
Tries #{{ task.tries }}: {{ task.last_error_body }} ({{ task.last_error_status_code }})

{% endfor %}

{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block content %}
You're about to follow {{ profile }}

<form method="POST" action="">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}"/>
<input type="hidden" name="profile" value="{{ profile }}">
<input type="submit" value="Follow" class="pure-button pure-button-success">
</form>

{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block title %}Followers - {{ config.NAME }}{% endblock %}
{% block header %}
{% endblock %}
{% block content %}
{% include "header.html" %}
<div class="content pure-u-1 pure-u-md-3-4" id="followers">
{% for follower in followers_data %}
{{ utils.display_actor_inline(follower, default_avatar=config.DEFAULT_ICON, size=80) }}
{% endfor %}
{{ utils.display_pagination(older_than, newer_than) }}
</div>
</div>
{% endblock %}
{% block links %}
<link rel="alternate" href="{{ config.BASE_URL }}/followers" type="application/activity+json">
{{ utils.display_pagination_links(older_than, newer_than) }}{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block title %}Following - {{ config.NAME }}{% endblock %}
{% block header %}
{% endblock %}
{% block content %}
{% include "header.html" %}
<div class="content pure-u-1 pure-u-md-3-4" id="following">
{% for (follow_id, follow) in following_data %}
{{ utils.display_actor_inline(follow, default_avatar=config.DEFAULT_ICON, size=80,actions={"unfollowbtn":True,"fid":follow_id}) }}
{% endfor %}
{{ utils.display_pagination(older_than, newer_than) }}
</div>
</div>
{% endblock %}
{% block links %}
<link rel="alternate" href="{{ config.BASE_URL }}/following" type="application/activity+json">
{{ utils.display_pagination_links(older_than, newer_than) }}{% endblock %}


code:
{% if not request.path.startswith('/admin') %}

<data class="u-photo" value="{{ config.ME.icon.url }}"></data>

{{ config.NAME }}

@{{ config.USERNAME }}@{{ config.DOMAIN }}

{{ config.SUMMARY | safe }}

Remote follow

Notes {{ notes_count }}

{% if session.logged_in %}
With replies {{ with_replies_count }}

{% endif %}

Liked {{ liked_count }}

Followers {{ followers_count }}

{% if not config.HIDE_FOLLOWING or session.logged_in %}
Following {{ following_count }}
{% endif %}

© {{ config.COPYRIGHT }} | Imprint | Privacy | Source

Powered by a fork of microblog.pub and the ActivityPub protocol

{% endif %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block header %}
<link rel="self" href="{{ config.ID }}/">
<link rel="hub" href="https://pubsubhubbub.superfeedr.com/">
<meta content="profile" property="og:type"/>
<meta content="{{ config.ID }}" property="og:url"/>
<meta content="{{ config.USERNAME }}" property="og:site_name"/>
<meta content="{{ config.USERNAME }}" property="og:title"/>
<meta content="{{ me.summary | html2plaintext }}" property="og:description"/>
<meta content="{{ me.icon.url }}" property="og:image"/>
<meta content="200" property="og:image:width"/>
<meta content="200" property="og:image:height"/>
<meta content="summary" property="twitter:card"/>
<meta content="{{config.USERNAME}}@{{ config.DOMAIN}}" property="profile:username"/>
{% endblock %}
{% block content %}
{% include "header.html" %}
<data class="p-name" value="{{config.USERNAME}} notes"></data>

{% if pinned %}

Pinned Posts

{% for item in pinned %}
{{ utils.display_note(item.activity.object, default_avatar=config.DEFAULT_ICON, meta=item.meta, no_color=True, index=True, limit=config.PREVIEW_LIMIT) }}
{% endfor %}

{% endif %}

Recent Posts

{% for item in outbox_data %}
{% if item | has_type('Announce') %}
{% if "actor" in item.meta %}
{% set boost_actor = item.meta.actor %}

{{ boost_actor.name }} boosted

{% endif %}
{% if item.meta.object %}
{{ utils.display_note(item.meta.object, default_avatar=config.DEFAULT_ICON, ui=True, meta={'actor':item.meta.object_actor,'boosted':item.remote_id}, index=True, limit=config.PREVIEW_LIMIT) }}
{% endif %}
{% elif item | has_type('Create') %}
{{ utils.display_note(item.activity.object, default_avatar=config.DEFAULT_ICON, meta=item.meta, no_color=True, index=True, limit=config.PREVIEW_LIMIT) }}
{% endif %}
{% endfor %}
{{ utils.display_pagination(older_than, newer_than) }}

{% endblock %}
{% block links %}<link rel="alternate" href="{{ config.ME.url | get_url }}" title="ActivityPub profile" type="application/activity+json">
<link rel="alternate" href="{{ config.ID }}/feed.json" type="application/json" title="JSON Feed">
<link rel="alternate" href="{{ config.ID }}/feed.rss" type="application/rss+xml" title="RSS">
<link rel="alternate" href="{{ config.ID }}/feed.atom" type="application/atom+xml" title="Atom 0.3">
{{ utils.display_pagination_links(older_than, newer_than) }}{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block header %}
{% endblock %}
{% block content %}

{% if client.logo %}

{% endif %}
{{ client.name }} wants you to login.
<form method="POST" action="{{ url_for('indieauth_flow') }}">
{% if scopes %}
Scopes:

{% for scope in scopes %}

<input type="checkbox" name="scopes" value="{{scope}}" id="scope-{{scope}}"> {{ scope }}

{% endfor %}

{% endif %}
<input type="hidden" name="redirect_uri" value="{{ redirect_uri }}">
<input type="hidden" name="state" value="{{ state }}">
<input type="hidden" name="client_id" value="{{ client_id }}">
<input type="hidden" name="me" value="{{ me }}">
<input type="hidden" name="response_type" value="{{ response_type }}">

<input type="submit" value="Login" class="pure-button pure-button-success">
</form>

{% endblock %}


code:
<!DOCTYPE HTML>

<meta charset="utf-8">
<meta http-equiv="x-ua-compatible" content="ie=edge">
<meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no">
<link rel="shortcut icon" type="image/x-icon" href="{{ config.FAVICON }}">
<title>{% block title %}{{ config.NAME }}{% endblock %}</title>
<link rel="stylesheet" href="/static/pure.css">
<link rel="stylesheet" href="/static/grids.css">
<link rel="stylesheet" href="/static/style.css">
<link rel="stylesheet" href="/static/fonts.css">
<link rel="stylesheet" href="/static/cookie.css">

<link rel="authorization_endpoint" href="{{ config.ID }}/indieauth">
<link rel="token_endpoint" href="{{ config.ID }}/token">
{% if not request.args.get("older_than") and not request.args.get("previous_than") %}<link rel="canonical" href="{{ config.BASE_URL }}{{ request.path }}">{% endif %}
{% block links %}{% endblock %}
{% block headers %}{% endblock %}

{% if logged_in %}

Admin

Public

New

Stream

Notifications

Following

Lookup

Logout

{% endif %}

{% block content %}{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block content %}
<data class="p-name" value="{{config.USERNAME}} liked"></data>
{% include "header.html" %}

Liked Posts

{% for item in liked %}
{% if item.meta.object %}
{{ utils.display_note(item.meta.object, default_avatar=config.DEFAULT_ICON,ui=True,meta={'actor': item.meta.object_actor,'liked': item.remote_id}, index=True, limit=config.PREVIEW_LIMIT) }}
{% endif %}
{% endfor %}
{{ utils.display_pagination(older_than, newer_than) }}

{% endblock %}
{% block links %}
<link rel="alternate" href="{{ config.BASE_URL }}/liked" type="application/activity+json">
{{ utils.display_pagination_links(older_than, newer_than) }}{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block title %}Login - {{ config.NAME }}{% endblock %}
{% include "header.html" %}
{% block content %}

Please enter your password!

<form id="login-form" method="POST" style="display:inline">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<input type="password" name="pass" placeholder="password">

{% if u2f_enabled %}
<input type="hidden" name="resp" id="sig-payload" value="">
{% else %}
{% endif %}
<input type="submit" value="Login" class="pure-button pure-button-success">
</form>

{% if u2f_enabled %}

{% endif %}
{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block title %}Lookup - {{ config.NAME }}{% endblock %}
{% block content %}
{% include "header.html" %}

Please enter the URL of a Mastodon/Pleroma profile or post.

<form id="lookup-form" method="POST">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<input type="text" name="url" value="{{url or '' }}" placeholder="https://url-of-a-user-or-a-note.tld" style="width:100%">

<input type="submit" value="Lookup" class="pure-button pure-button-success">
</form>

{% if data %}
{% set data = data.to_dict() %}

{% if data | has_actor_type %}
{{ utils.display_actor_inline(data, default_avatar=config.DEFAULT_ICON, size=80,actions={"followbtn":True}) }}
{% elif data | has_type('Create') %}
{{ utils.display_note(data.object, default_avatar=config.DEFAULT_ICON, ui=True) }}
{% elif data | has_type(['Note', 'Article', 'Video']) %}
{{ utils.display_note(data, default_avatar=config.DEFAULT_ICON, ui=True) }}
{% elif data | has_type('Announce') %}
{% set boost_actor = meta.actor %}

{{ boost_actor.name }} boosted

{{ utils.display_note(meta.object, default_avatar=config.DEFAULT_ICON, ui=False, meta={'actor': meta.object_actor}) }}
{% endif %}

{% endif %}
{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block title %}New - {{ config.NAME }}{% endblock %}
{% block headers %}
{% endblock %}
{% block content %}
{% include "header.html" %}
{% if thread %}

Replying to {{ content }}

{{ utils.display_thread(thread) }}

{% endif %}

{% if thread %}

New reply

{% else %}
{% if request.args.get("question") == "1" %}

New question make it a note?

{% else %}

New note make it a question?

{% endif %}
{% endif %}
<form action="/api/new_{% if request.args.get("question") == "1" %}question{%else%}note{%endif%}" method="POST" enctype="multipart/form-data">
<input type="hidden" name="redirect" value="/">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
{% if reply %}<input type="hidden" name="reply" value="{{reply}}">{% endif %}

{% for emoji in emojis %}
{{ emoji }}
{% endfor %}

<textarea name="content" autofocus="autofocus" designMode="on" style="width:100%;height:500px;resize:none">{{ content }}</textarea>

<input type="file" name="file" style="color:#FFFFFF">
{% if request.args.get("question") == "1" %}

Open for: <select name="open_for" style="color:#000000">
<option value="30">30 minutes</option>
<option value="60">1 hour</option>
<option value="360">6 hour</option>
<option value="1440" selected>1 day</option>
<option value="4320">3 days</option>
<option value="10080">7 days</option>
</select>

<input type="hidden" name="of" value="oneOf" />

<select name="of" style="color:#000000">
<option value="oneOf">Single choice</option>
<option value="anyOf">Multiple choices</option>
</select>

{% for i in range(4) %}

<input type="text" name="answer{{i}}" placeholder="Answer #{{i+1}}">

{% endfor %}

{% endif %}
<input type="submit" value="Post" class="pure-button pure-button-success" style="float:right">

</form>

{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block title %}{{ config.NAME }}: "{{ note.activity.object.content | html2plaintext | truncate(50) }}"{% endblock %}
{% block header %}
<link rel="self" href="{{ config.ID }}/">
<link rel="hub" href="https://pubsubhubbub.superfeedr.com/">
<meta content="article" property="og:type"/>
<meta content="{{ note.activity.object.url }}" property="og:url"/>
<meta content="{{ config.USERNAME }}" property="og:site_name"/>
<meta content="{{ config.USERNAME }}" property="og:title"/>
<meta content="{{ note.activity.object.content | html2plaintext | truncate(50) }}" property="og:description"/>
<meta content="{{ me.icon.url }}" property="og:image"/>
<meta content="200" property="og:image:width"/>
<meta content="200" property="og:image:height"/>
<meta content="summary" property="twitter:card"/>
<meta content="{{config.USERNAME}}@{{ config.DOMAIN}}" property="profile:username"/>
{% endblock %}
{% block content %}
{% include "header.html" %}
{% if request.path.startswith('/admin') %}

Thread

{{ utils.display_thread(thread, default_avatar=config.DEFAULT_ICON, likes=likes, shares=shares) }}

{% else %}

Post

{{ utils.display_thread(thread, default_avatar=config.DEFAULT_ICON, likes=likes, shares=shares) }}

{% endif %}
{% endblock %}
{% block links %}<link rel="alternate" href="{{ note.activity.object.id}}" type="application/activity+json">{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block title %}{{ config.NAME }}: "{{ note.activity.object.content | html2plaintext | truncate(50) }}"{% endblock %}
{% block header %}
<meta content="article" property="og:type" />
<meta content="{{ note.activity.object.url }}" property="og:url" />

<meta content="{{ config.USERNAME }}" property="og:site_name" />
<meta content="{{ config.USERNAME }}" property="og:title" />
<meta content="{{ note.activity.object.content | html2plaintext | truncate(50) }}" property="og:description" />
<meta content="{{ me.icon.url }}" property="og:image" />
<meta content="200" property="og:image:width" />
<meta content="200" property="og:image:height" />
<meta content="summary" property="twitter:card" />
{% endblock %}
{% block content %}

{% include "header.html" %}

{{ thread | remove_mongo_id | tojson(indent=4) }}

{% endblock %}
{% block links %}<link rel="alternate" href="{{ note.activity.object.id}}" type="application/activity+json">{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block content %}
{% include "header.html" %}

Please insert your username to proceed to follow!

<form method="POST" action="" style="display:inline">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<input type="text" name="profile" placeholder="you@your-instance">

<input type="submit" value="Proceed to remote follow" class="pure-button pure-button-success">
</form>

{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block title %}{% if request.path == url_for('admin_stream') %}Stream{% else %}Notifications{% endif %} - {{ config.NAME }}{% endblock %}
{% block content %}
{% include "header.html" %}

Recent Activities

{% for item in inbox_data %}
{% if 'actor' in item.meta %}
{% if item | has_type('Create') %}
{{ utils.display_note(item.activity.object, default_avatar=config.DEFAULT_ICON, ui=True, meta=item.meta) }}
{% else %}
{% if item | has_type('Announce') %}
{% set boost_actor = item.meta.actor %}
{% if boost_actor %}

{{ boost_actor.name or boost_actor.preferredUsername }} boosted

{% endif %}
{% if item.meta.object %}
{{ utils.display_note(item.meta.object, default_avatar=config.DEFAULT_ICON, ui=True) }}
{% endif %}
{% endif %}
{% if item | has_type('Like') %}
{% set boost_actor = item.meta.actor %}

{{ boost_actor.name or boost_actor.preferredUsername }} liked

{% if item.meta.object %}
{{ utils.display_note(item.meta.object, default_avatar=config.DEFAULT_ICON, ui=False, meta={'actor': item.meta.object_actor}) }}
{% endif %}
{% endif %}
{% if item | has_type('Follow') %}

new follower

{{ utils.display_actor_inline(item.meta.actor, default_avatar=config.DEFAULT_ICON, size=50) }}

{% elif item | has_type('Accept') %}

you started following

{{ utils.display_actor_inline(item.meta.actor, default_avatar=config.DEFAULT_ICON, size=50) }}

{% elif item | has_type('Undo') %}

unfollowed you

{{ utils.display_actor_inline(item.meta.actor, default_avatar=config.DEFAULT_ICON, size=50) }}

{% else %}
{% endif %}
{% endif %}
{% endif %}
{% endfor %}
{{ utils.display_pagination(older_than, newer_than) }}

{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block title %}{% if request.path == url_for('admin_stream') %}Stream{% else %}Notifications{% endif %} - {{ config.NAME }}{% endblock %}
{% block content %}

{% include "header.html" %}

	{% for item in inbox_data %}

{{ item |remove_mongo_id|tojson(indent=4) }}

	{% endfor %}

    {{ utils.display_pagination(older_than, newer_than) }}

{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block title %}#{{ tag }} - {{ config.NAME }}{% endblock %}
{% block header %}
<link rel="self" href="{{ config.ID }}/">
<link rel="hub" href="https://pubsubhubbub.superfeedr.com/">
<meta content="article" property="og:type"/>
<meta content="{{ config.ID }}" property="og:url"/>
<meta content="{{ config.USERNAME }}" property="og:site_name"/>
<meta content="{{ config.USERNAME }}" property="og:title"/>
<meta content="{{ me.summary | html2plaintext }}" property="og:description"/>
<meta content="{{ me.icon.url }}" property="og:image"/>
<meta content="200" property="og:image:width"/>
<meta content="200" property="og:image:height"/>
<meta content="summary" property="twitter:card"/>
<meta content="{{config.USERNAME}}@{{ config.DOMAIN}}" property="profile:username"/>
{% endblock %}
{% block content %}
<data class="p-name" value="#{{tag}}"></data>
{% include "header.html" %}

#{{ tag }}

{% for item in outbox_data %}
{{ utils.display_note(item.activity.object, default_avatar=config.DEFAULT_ICON, meta=item.meta, index=True, limit=config.PREVIEW_LIMIT) }}
{% endfor %}

{% endblock %}
{% block links %}
<link rel="alternate" href="{{ config.BASE_URL }}/tags/{{tag}}" type="application/activity+json">
{% endblock %}


code:
{% extends "layout.html" %}
{% import 'utils.html' as utils %}
{% block header %}
{% endblock %}
{% block content %}
<div style="width:90%;max-width: 720px;margin:40px auto;">
{% if session.logged_in %}logged{% else%}not logged{%endif%}
<form method="POST" id="register-form">
<input type="hidden" id="register-payload" name="resp" value="">
<input type="submit" value="login">
</form>
</div>
<script>
var p= {{ payload | tojson }};
//setTimeout(function() {
u2f.register(p.appId, p.registerRequests, p.registeredKeys, function(resp) {
if (resp.errorCode) {
console.log(resp)
} else {
document.getElementById('register-payload').value = JSON.stringify(resp)
document.getElementById('register-form').submit()
}

})
//}, 1000);
</script>
{% endblock %}


code:
{% macro display_actor_inline(follower, default_avatar, size=50,actions={}) -%}
{% if follower and follower.id %}

{% if not follower.icon %}

{% else %}
{% endif %}

{% if actions.followbtn %}

<form action="/api/follow" class="action-form" method="POST">
<input type="hidden" name="redirect" value="{{ request.path }}"/>
<input type="hidden" name="actor" value="{{ follower.id }}"/>
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}"/>
<button type="submit" class="bar-item pure-button">follow</button>
</form>

{% endif %}
{% if actions.unfollowbtn %}

<form action="/api/undo" class="action-form"  method="POST">
<input type="hidden" name="redirect" value="{{ request.path }}"/>
<input type="hidden" name="id" value="{{ fid }}"/>
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}"/>
<button type="submit" class="bar-item pure-button">unfollow</button>
</form>

{% endif %}

{{ follower.name or follower.preferredUsername }}

@{{ follower.preferredUsername }}@{{ follower | url_or_id | get_url | domain }}

{% endif %}
{%- endmacro %}

{% macro display_note(obj, default_avatar, perma=False, ui=False, likes=[], shares=[], meta={}, no_color=False, index=False, limit=0) -%}

{% if meta.actor %}
    {% set actor = meta.actor %}
{% else %}
    {% set actor = obj.attributedTo | get_actor  %}
{% endif %}

{% if session.logged_in %}
    {% set perma_id = obj.id | permalink_id %}

    {% if request.args.get('older_than') %}
        {% set redir = request.path + "?older_than=" + request.args.get('older_than') + "#activity-" + perma_id %}
    {% elif request.args.get('newer_than') %}
        {% set redir = request.path + "?newer_than=" + request.args.get('newer_than') + "#activity-" + perma_id %}
    {% else %}
    {% set redir = request.path + "#activity-" + perma_id %}
    {% endif %}

    {% set aid = obj.id | quote_plus %}
{% endif %}

{{ actor.name or actor.preferredUsername }}
@{% if not no_color and obj.id | is_from_outbox %}{{ actor.preferredUsername }}{% else %}{{ actor.preferredUsername }}{% endif %}@{% if not no_color and obj.id | is_from_outbox %}{{ actor | url_or_id | get_url | domain }}{% else %}{{ actor | url_or_id | get_url | domain }}{% endif %} published
<time  title="{{ obj.published }}" datetime="{{ obj.published }}">{{ obj.published | format_timeago }}</time>

	{% if obj.summary %}
{{ obj.summary | clean }}
{% endif %}
    {% if obj | has_type('Video') %}

    <video controls preload="metadata"  src="{{ obj.url }}" width="480">
    </video>

    {% endif %}

    {% if obj | has_type('Article') %}
        {{ obj.name }} {{ obj | url_or_id | get_url }}
    {% elif obj | has_type('Question') %}
	{% if index and limit != 0 and obj.content|length > limit %}
            {{ obj.content | truncate(limit) | clean | safe }}
            Read more.
        {% else %}
            {{ obj.content | clean | safe }}
        {% endif %}

        {% if obj.id | is_from_outbox or (meta.question_replies and (obj.closed or meta.voted_for)) %}

        {% set total_votes = meta.question_replies %}

        {% for oneOf in obj.oneOf %}
            {% set pct = 0 %}
            {% if total_votes > 0 %}
                {% set cnt = oneOf.name | get_answer_count(meta) %}
                {% set pct = cnt * 100.0 / total_votes %}
            {% endif %}

                    {{ '%0.0f'| format(pct) }}%
                    {{ oneOf.name }}

        {% endfor %}

    {% if obj.closed or obj.endTime | gtnow %}
        Ended {{ obj.endTime | format_timeago }} with {{ total_votes }} vote{% if total_votes | gtone %}s{% endif %}.
    {% else %}
        Ends {{ obj.endTime | format_timeago }} ({{ total_votes }} vote{% if total_votes | gtone %}s{% endif %} as of now).
    {% endif %}

    {% else %}

    {% for oneOf in obj.oneOf %}

{% if not meta.voted_for and not obj.endTime | gtnow %}
<form action="/api/vote" class="action-form"  method="POST">
<input type="hidden" name="redirect" value="{{ redir }}">
<input type="hidden" name="id" value="{{ obj.id }}">
<input type="hidden" name="choice" value="{{ oneOf.name }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="bar-item">vote</button>
</form>
{% else %}
    ???
{% endif %}
{{ oneOf.name }} {% if oneOf.name == meta.voted_for %}(your vote){% endif %}

    {% endfor %}

{% if obj.endTime | gtnow %}This question ended {{ obj.endTime | format_timeago }}.

    {% else %}This question ends {{ obj.endTime | format_timeago }}{% endif %}

    {% endif %}

    {% else %}
	{% if index and limit != 0 and obj.content|length > limit %}
            {{ obj.content | truncate(limit) | clean | safe }}
            Read more.
        {% else %}
            {{ obj.content | clean | safe }}
        {% endif %}
    {% endif %}

	{% if obj.attachment and obj | has_type('Note') %}

	{% if obj.attachment | not_only_imgs %}

Attachments

	{% endif %}
	{% for a in obj.attachment %}
    {% if (a.mediaType and a.mediaType.startswith("image/")) or (a.type and a.type == 'Image') %}

    {% elif (a.mediaType and a.mediaType.startswith("video/")) %}

<video controls preload="metadata"  src="{{ a.url }}" width="480"></video>

	{% else %}

{% if a.filename %}{{ a.filename }}{% else %}{{ a.url }}{% endif %}

	{% endif  %}
	{% endfor %}
	{% if obj.attachment | not_only_imgs %}

	{% endif %}

	{% endif %}

{% if meta and meta.og_metadata and obj | has_type('Note') %}
{% for og in meta.og_metadata %}
{% if og.url %}

{{ og.title }}

{{ og.description | truncate(80) }}

{{ og.site_name }}

{% endif %}
{% endfor %}
{% endif %}

{% if perma %}{{ obj.published | format_time }}{% endif %}

{% if perma %}
{% if not (obj.id | is_from_outbox) %}
permalink
{% endif %}
{% else %}
permalink
{% endif %}
{% if meta.count_reply and obj.id | is_from_outbox %}{{ meta.count_reply }} replies
{% elif meta.count_reply and session.logged_in %}
{{ meta.count_reply }} replies{% endif %}

{% if not perma and meta.count_boost and obj.id | is_from_outbox %}{{ meta.count_boost }} boosts{% endif %}
{% if not perma and meta.count_like and obj.id | is_from_outbox %}{{ meta.count_like }} likes{% endif %}

{% if session.logged_in %}
{% if ui %}
{% if meta.boosted %}
<form action="/api/undo" class="action-form" method="POST">
<input type="hidden" name="redirect" value="{{ redir }}">
<input type="hidden" name="id" value="{{ meta.boosted }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="bar-item pure-button">unboost</button>
</form>
{% else %}
<form action="/api/boost" class="action-form" method="POST">
<input type="hidden" name="redirect" value="{{ redir }}">
<input type="hidden" name="id" value="{{ obj.id }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="bar-item pure-button">boost</button>
</form>
{% endif %}

{% if meta.liked %}
<form action="/api/undo" class="action-form" method="POST">
<input type="hidden" name="redirect" value="{{ redir }}">
<input type="hidden" name="id" value="{{ meta.liked }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="bar-item pure-button">unlike</button>
</form>
{% else %}
<form action="/api/like" class="action-form" method="POST">
<input type="hidden" name="redirect" value="{{ redir }}">
<input type="hidden" name="id" value="{{ obj.id }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="bar-item pure-button">like</button>
</form>
{% endif %}
{% endif %}
{% if obj.id | is_from_outbox %}
<form action="/api/note/delete" class="action-form" method="POST">
<input type="hidden" name="redirect" value="{{ redir }}">
<input type="hidden" name="id" value="{{ obj.id }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="bar-item pure-button">delete</button>
</form>
{% if meta.pinned %}
<form action="/api/note/unpin" class="action-form" method="POST">
<input type="hidden" name="redirect" value="{{ redir }}">
<input type="hidden" name="id" value="{{ obj.id }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="bar-item pure-button">unpin</button>
</form>
{% else %}
<form action="/api/note/pin" class="action-form" method="POST">
<input type="hidden" name="redirect" value="{{ redir }}">
<input type="hidden" name="id" value="{{ obj.id }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="bar-item pure-button">pin</button>
</form>
{% endif %}

{% else %}
<form action="/api/block" class="action-form" method="POST">
<input type="hidden" name="redirect" value="{{ redir }}">
<input type="hidden" name="actor" value="{{ actor.id }}">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="bar-item pure-button">block</button>
</form>
{% endif %}
reply
{% endif %}

{% if likes or shares %}

{% if likes %}

{{ likes|length }} likes
{% for like in likes %}
{{ display_actor_inline(like) }}
{% endfor %}

{% endif %}
{% if shares %}

{{ shares|length }} boosts
{% for boost in shares %}
{{ display_actor_inline(boost) }}
{% endfor %}

{% endif %}

{% endif %}

{%- endmacro %}

{% macro display_thread(thread, default_avatar, likes=[], shares=[]) -%}
{% for reply in thread %}
{% if reply._requested %}
{{ display_note(reply.activity.object, default_avatar=default_avatar, perma=True, ui=False, likes=likes, shares=shares, meta=reply.meta) }}
{% else %}
{{ display_note(reply.activity.object, default_avatar=default_avatar, perma=False, ui=True, meta=reply.meta) }}
{% endif %}
{% endfor %}
{% endmacro -%}

{% macro display_pagination(older_than, newer_than) -%}

    {% if older_than %}
    🡨 Older
    {% endif %}
    {% if newer_than %}
    Newer 🡪
    {% endif %}

{% endmacro -%}

{% macro display_pagination_links(older_than, newer_than) -%}
{% if older_than %}<link rel="next" href="{{ config.BASE_URL }}{{ request.path }}?older_than={{older_than}}">{% endif %}
{% if newer_than %}<link rel="prev" href="{{ config.BASE_URL }}{{ request.path }}?newer_than={{newer_than}}">{% endif %}
{% endmacro -%}


code:
import os
import time
from typing import List
from typing import Tuple

import requests
from html2text import html2text
from little_boxes.collection import parse_collection

def resp2plaintext(resp):
    """Convert the body of a requests reponse to plain text in order to make basic assertions."""
    return html2text(resp.text)

class Instance(object):
    """Test instance wrapper."""

    def __init__(self, name, host_url, docker_url=None):
        self.host_url = host_url
        self.docker_url = docker_url or host_url
        self._create_delay = 10
        with open(
            os.path.join(
                os.path.dirname(os.path.abspath(__file__)),
                f"fixtures/{name}/config/admin_api_key.key",
            )
        ) as f:
            api_key = f.read()
        self._auth_headers = {"Authorization": f"Bearer {api_key}"}

    def _do_req(self, url):
        """Used to parse collection."""
        url = url.replace(self.docker_url, self.host_url)
        resp = requests.get(url, headers={"Accept": "application/activity+json"})
        resp.raise_for_status()
        return resp.json()

    def _parse_collection(self, payload=None, url=None):
        """Parses a collection (go through all the pages)."""
        return parse_collection(url=url, payload=payload, fetcher=self._do_req)

    def ping(self):
        """Ensures the homepage is reachable."""
        resp = requests.get(f"{self.host_url}/")
        resp.raise_for_status()
        assert resp.status_code == 200

    def debug(self):
        """Returns the debug infos (number of items in the inbox/outbox."""
        resp = requests.get(
            f"{self.host_url}/api/debug",
            headers={**self._auth_headers, "Accept": "application/json"},
        )
        resp.raise_for_status()

        return resp.json()

    def drop_db(self):
        """Drops the MongoDB DB."""
        resp = requests.delete(
            f"{self.host_url}/api/debug",
            headers={**self._auth_headers, "Accept": "application/json"},
        )
        resp.raise_for_status()

        return resp.json()

    def block(self, actor_url) -> None:
        """Blocks an actor."""
        # Instance1 follows instance2
        resp = requests.post(
            f"{self.host_url}/api/block",
            params={"actor": actor_url},
            headers=self._auth_headers,
        )
        assert resp.status_code == 201

        # We need to wait for the Follow/Accept dance
        time.sleep(self._create_delay / 2)
        return resp.json().get("activity")

    def follow(self, instance: "Instance") -> str:
        """Follows another instance."""
        # Instance1 follows instance2
        resp = requests.post(
            f"{self.host_url}/api/follow",
            json={"actor": instance.docker_url},
            headers=self._auth_headers,
        )
        assert resp.status_code == 201

        # We need to wait for the Follow/Accept dance
        time.sleep(self._create_delay)
        return resp.json().get("activity")

    def new_note(self, content, reply=None) -> str:
        """Creates a new note."""
        params = {"content": content}
        if reply:
            params["reply"] = reply

        resp = requests.post(
            f"{self.host_url}/api/new_note", json=params, headers=self._auth_headers
        )
        assert resp.status_code == 201

        time.sleep(self._create_delay)
        return resp.json().get("activity")

    def boost(self, oid: str) -> str:
        """Creates an Announce activity."""
        resp = requests.post(
            f"{self.host_url}/api/boost", json={"id": oid}, headers=self._auth_headers
        )
        assert resp.status_code == 201

        time.sleep(self._create_delay)
        return resp.json().get("activity")

    def like(self, oid: str) -> str:
        """Creates a Like activity."""
        resp = requests.post(
            f"{self.host_url}/api/like", json={"id": oid}, headers=self._auth_headers
        )
        assert resp.status_code == 201

        time.sleep(self._create_delay)
        return resp.json().get("activity")

    def delete(self, oid: str) -> str:
        """Creates a Delete activity."""
        resp = requests.post(
            f"{self.host_url}/api/note/delete",
            json={"id": oid},
            headers=self._auth_headers,
        )
        assert resp.status_code == 201

        time.sleep(self._create_delay)
        return resp.json().get("activity")

    def undo(self, oid: str) -> str:
        """Creates a Undo activity."""
        resp = requests.post(
            f"{self.host_url}/api/undo", json={"id": oid}, headers=self._auth_headers
        )
        assert resp.status_code == 201

        # We need to wait for the Follow/Accept dance
        time.sleep(self._create_delay)
        return resp.json().get("activity")

    def followers(self) -> List[str]:
        """Parses the followers collection."""
        resp = requests.get(
            f"{self.host_url}/followers",
            headers={"Accept": "application/activity+json"},
        )
        resp.raise_for_status()

        data = resp.json()

        return self._parse_collection(payload=data)

    def following(self):
        """Parses the following collection."""
        resp = requests.get(
            f"{self.host_url}/following",
            headers={"Accept": "application/activity+json"},
        )
        resp.raise_for_status()

        data = resp.json()

        return self._parse_collection(payload=data)

    def outbox(self):
        """Returns the instance outbox."""
        resp = requests.get(
            f"{self.host_url}/following",
            headers={"Accept": "application/activity+json"},
        )
        resp.raise_for_status()
        return resp.json()

    def outbox_get(self, aid):
        """Fetches a specific item from the instance outbox."""
        resp = requests.get(
            aid.replace(self.docker_url, self.host_url),
            headers={"Accept": "application/activity+json"},
        )
        resp.raise_for_status()
        return resp.json()

    def stream_jsonfeed(self):
        """Returns the "stream"'s JSON feed."""
        resp = requests.get(
            f"{self.host_url}/api/stream",
            headers={**self._auth_headers, "Accept": "application/json"},
        )
        resp.raise_for_status()
        return resp.json()

def _instances() -> Tuple[Instance, Instance]:
    """Initializes the client for the two test instances."""
    instance1 = Instance("instance1", "http://docker:5006", "http://instance1_web:5005")
    instance1.ping()

    instance2 = Instance("instance2", "http://docker:5007", "http://instance2_web:5005")
    instance2.ping()

    # Return the DB
    instance1.drop_db()
    instance2.drop_db()

    return instance1, instance2

def test_follow() -> None:
    """instance1 follows instance2."""
    instance1, instance2 = _instances()
    # Instance1 follows instance2
    instance1.follow(instance2)
    instance1_debug = instance1.debug()
    assert instance1_debug["inbox"] == 1  # An Accept activity should be there
    assert instance1_debug["outbox"] == 1  # We've sent a Follow activity

    instance2_debug = instance2.debug()
    assert instance2_debug["inbox"] == 1  # An Follow activity should be there
    assert instance2_debug["outbox"] == 1  # We've sent a Accept activity

    assert instance2.followers() == [instance1.docker_url]
    assert instance1.following() == [instance2.docker_url]

def test_follow_unfollow():
    """instance1 follows instance2, then unfollows it."""
    instance1, instance2 = _instances()
    # Instance1 follows instance2
    follow_id = instance1.follow(instance2)
    instance1_debug = instance1.debug()
    assert instance1_debug["inbox"] == 1  # An Accept activity should be there
    assert instance1_debug["outbox"] == 1  # We've sent a Follow activity

    instance2_debug = instance2.debug()
    assert instance2_debug["inbox"] == 1  # An Follow activity should be there
    assert instance2_debug["outbox"] == 1  # We've sent a Accept activity

    assert instance2.followers() == [instance1.docker_url]
    assert instance1.following() == [instance2.docker_url]

    instance1.undo(follow_id)

    assert instance2.followers() == []
    assert instance1.following() == []

    instance1_debug = instance1.debug()
    assert instance1_debug["inbox"] == 1  # An Accept activity should be there
    assert instance1_debug["outbox"] == 2  # We've sent a Follow and a Undo activity

    instance2_debug = instance2.debug()
    assert instance2_debug["inbox"] == 2  # An Follow and Undo activity should be there
    assert instance2_debug["outbox"] == 1  # We've sent a Accept activity

def test_post_content():
    """Instances follow each other, and instance1 creates a note."""
    instance1, instance2 = _instances()
    # Instance1 follows instance2
    instance1.follow(instance2)
    instance2.follow(instance1)

    inbox_stream = instance2.stream_jsonfeed()
    assert len(inbox_stream["items"]) == 0

    create_id = instance1.new_note("hello")
    instance2_debug = instance2.debug()
    assert (
        instance2_debug["inbox"] == 3
    )  # An Follow, Accept and Create activity should be there
    assert instance2_debug["outbox"] == 2  # We've sent a Accept and a Follow  activity

    # Ensure the post is visible in instance2's stream
    inbox_stream = instance2.stream_jsonfeed()
    assert len(inbox_stream["items"]) == 1
    assert inbox_stream["items"][0]["id"] == create_id

def test_block_and_post_content():
    """Instances follow each other, instance2 blocks instance1, instance1 creates a new note."""
    instance1, instance2 = _instances()
    # Instance1 follows instance2
    instance1.follow(instance2)
    instance2.follow(instance1)

    inbox_stream = instance2.stream_jsonfeed()
    assert len(inbox_stream["items"]) == 0

    instance2.block(instance1.docker_url)

    instance1.new_note("hello")
    instance2_debug = instance2.debug()
    assert (
        instance2_debug["inbox"] == 2
    )  # An Follow, Accept activity should be there, Create should have been dropped
    assert (
        instance2_debug["outbox"] == 3
    )  # We've sent a Accept and a Follow  activity + the Block activity

    # Ensure the post is not visible in instance2's stream
    inbox_stream = instance2.stream_jsonfeed()
    assert len(inbox_stream["items"]) == 0

def test_post_content_and_delete():
    """Instances follow each other, instance1 creates a new note, then deletes it."""
    instance1, instance2 = _instances()
    # Instance1 follows instance2
    instance1.follow(instance2)
    instance2.follow(instance1)

    inbox_stream = instance2.stream_jsonfeed()
    assert len(inbox_stream["items"]) == 0

    create_id = instance1.new_note("hello")
    instance2_debug = instance2.debug()
    assert (
        instance2_debug["inbox"] == 3
    )  # An Follow, Accept and Create activity should be there
    assert instance2_debug["outbox"] == 2  # We've sent a Accept and a Follow  activity

    # Ensure the post is visible in instance2's stream
    inbox_stream = instance2.stream_jsonfeed()
    assert len(inbox_stream["items"]) == 1
    assert inbox_stream["items"][0]["id"] == create_id

    instance1.delete(f"{create_id}/activity")
    instance2_debug = instance2.debug()
    assert (
        instance2_debug["inbox"] == 4
    )  # An Follow, Accept and Create and Delete activity should be there
    assert instance2_debug["outbox"] == 2  # We've sent a Accept and a Follow  activity

    # Ensure the post has been delete from instance2's stream
    inbox_stream = instance2.stream_jsonfeed()
    assert len(inbox_stream["items"]) == 0

def test_post_content_and_like():
    """Instances follow each other, instance1 creates a new note, instance2 likes it."""
    instance1, instance2 = _instances()
    # Instance1 follows instance2
    instance1.follow(instance2)
    instance2.follow(instance1)

    create_id = instance1.new_note("hello")

    # Ensure the post is visible in instance2's stream
    inbox_stream = instance2.stream_jsonfeed()
    assert len(inbox_stream["items"]) == 1
    assert inbox_stream["items"][0]["id"] == create_id

    # Now, instance2 like the note
    like_id = instance2.like(f"{create_id}/activity")

    instance1_debug = instance1.debug()
    assert instance1_debug["inbox"] == 3  # Follow, Accept and Like
    assert instance1_debug["outbox"] == 3  # Folllow, Accept, and Create

    note = instance1.outbox_get(f"{create_id}/activity")
    assert "likes" in note
    assert note["likes"]["totalItems"] == 1
    likes = instance1._parse_collection(url=note["likes"]["first"])
    assert len(likes) == 1
    assert likes[0]["id"] == like_id

def test_post_content_and_like_unlike() -> None:
    """Instances follow each other, instance1 creates a new note, instance2 likes it, then unlikes it."""
    instance1, instance2 = _instances()
    # Instance1 follows instance2
    instance1.follow(instance2)
    instance2.follow(instance1)

    create_id = instance1.new_note("hello")

    # Ensure the post is visible in instance2's stream
    inbox_stream = instance2.stream_jsonfeed()
    assert len(inbox_stream["items"]) == 1
    assert inbox_stream["items"][0]["id"] == create_id

    # Now, instance2 like the note
    like_id = instance2.like(f"{create_id}/activity")

    instance1_debug = instance1.debug()
    assert instance1_debug["inbox"] == 3  # Follow, Accept and Like
    assert instance1_debug["outbox"] == 3  # Folllow, Accept, and Create

    note = instance1.outbox_get(f"{create_id}/activity")
    assert "likes" in note
    assert note["likes"]["totalItems"] == 1
    likes = instance1._parse_collection(url=note["likes"]["first"])
    assert len(likes) == 1
    assert likes[0]["id"] == like_id

    instance2.undo(like_id)

    instance1_debug = instance1.debug()
    assert instance1_debug["inbox"] == 4  # Follow, Accept and Like and Undo
    assert instance1_debug["outbox"] == 3  # Folllow, Accept, and Create

    note = instance1.outbox_get(f"{create_id}/activity")
    assert "likes" in note
    assert note["likes"]["totalItems"] == 0

def test_post_content_and_boost() -> None:
    """Instances follow each other, instance1 creates a new note, instance2 "boost" it."""
    instance1, instance2 = _instances()
    # Instance1 follows instance2
    instance1.follow(instance2)
    instance2.follow(instance1)

    create_id = instance1.new_note("hello")

    # Ensure the post is visible in instance2's stream
    inbox_stream = instance2.stream_jsonfeed()
    assert len(inbox_stream["items"]) == 1
    assert inbox_stream["items"][0]["id"] == create_id

    # Now, instance2 like the note
    boost_id = instance2.boost(f"{create_id}/activity")

    instance1_debug = instance1.debug()
    assert instance1_debug["inbox"] == 3  # Follow, Accept and Announce
    assert instance1_debug["outbox"] == 3  # Folllow, Accept, and Create

    note = instance1.outbox_get(f"{create_id}/activity")
    assert "shares" in note
    assert note["shares"]["totalItems"] == 1
    shares = instance1._parse_collection(url=note["shares"]["first"])
    assert len(shares) == 1
    assert shares[0]["id"] == boost_id

def test_post_content_and_boost_unboost() -> None:
    """Instances follow each other, instance1 creates a new note, instance2 "boost" it, then "unboost" it."""
    instance1, instance2 = _instances()
    # Instance1 follows instance2
    instance1.follow(instance2)
    instance2.follow(instance1)

    create_id = instance1.new_note("hello")

    # Ensure the post is visible in instance2's stream
    inbox_stream = instance2.stream_jsonfeed()
    assert len(inbox_stream["items"]) == 1
    assert inbox_stream["items"][0]["id"] == create_id

    # Now, instance2 like the note
    boost_id = instance2.boost(f"{create_id}/activity")

    instance1_debug = instance1.debug()
    assert instance1_debug["inbox"] == 3  # Follow, Accept and Announce
    assert instance1_debug["outbox"] == 3  # Folllow, Accept, and Create

    note = instance1.outbox_get(f"{create_id}/activity")
    assert "shares" in note
    assert note["shares"]["totalItems"] == 1
    shares = instance1._parse_collection(url=note["shares"]["first"])
    assert len(shares) == 1
    assert shares[0]["id"] == boost_id

    instance2.undo(boost_id)

    instance1_debug = instance1.debug()
    assert instance1_debug["inbox"] == 4  # Follow, Accept and Announce and Undo
    assert instance1_debug["outbox"] == 3  # Folllow, Accept, and Create

    note = instance1.outbox_get(f"{create_id}/activity")
    assert "shares" in note
    assert note["shares"]["totalItems"] == 0

def test_post_content_and_post_reply() -> None:
    """Instances follow each other, instance1 creates a new note, instance2 replies to it."""
    instance1, instance2 = _instances()
    # Instance1 follows instance2
    instance1.follow(instance2)
    instance2.follow(instance1)

    inbox_stream = instance2.stream_jsonfeed()
    assert len(inbox_stream["items"]) == 0

    instance1_create_id = instance1.new_note("hello")
    instance2_debug = instance2.debug()
    assert (
        instance2_debug["inbox"] == 3
    )  # An Follow, Accept and Create activity should be there
    assert instance2_debug["outbox"] == 2  # We've sent a Accept and a Follow  activity

    # Ensure the post is visible in instance2's stream
    instance2_inbox_stream = instance2.stream_jsonfeed()
    assert len(instance2_inbox_stream["items"]) == 1
    assert instance2_inbox_stream["items"][0]["id"] == instance1_create_id

    instance2_create_id = instance2.new_note(
        f"hey @instance1@{instance1.docker_url}",
        reply=f"{instance1_create_id}/activity",
    )
    instance2_debug = instance2.debug()
    assert (
        instance2_debug["inbox"] == 3
    )  # An Follow, Accept and Create activity should be there
    assert (
        instance2_debug["outbox"] == 3
    )  # We've sent a Accept and a Follow and a Create  activity

    instance1_debug = instance1.debug()
    assert (
        instance1_debug["inbox"] == 3
    )  # An Follow, Accept and Create activity should be there
    assert (
        instance1_debug["outbox"] == 3
    )  # We've sent a Accept and a Follow and a Create  activity

    instance1_inbox_stream = instance1.stream_jsonfeed()
    assert len(instance1_inbox_stream["items"]) == 1
    assert instance1_inbox_stream["items"][0]["id"] == instance2_create_id

    instance1_note = instance1.outbox_get(f"{instance1_create_id}/activity")
    assert "replies" in instance1_note
    assert instance1_note["replies"]["totalItems"] == 1
    replies = instance1._parse_collection(url=instance1_note["replies"]["first"])
    assert len(replies) == 1
    assert replies[0]["id"] == f"{instance2_create_id}/activity"

def test_post_content_and_post_reply_and_delete() -> None:
    """Instances follow each other, instance1 creates a new note, instance2 replies to it, then deletes its reply."""
    instance1, instance2 = _instances()
    # Instance1 follows instance2
    instance1.follow(instance2)
    instance2.follow(instance1)

    inbox_stream = instance2.stream_jsonfeed()
    assert len(inbox_stream["items"]) == 0

    instance1_create_id = instance1.new_note("hello")
    instance2_debug = instance2.debug()
    assert (
        instance2_debug["inbox"] == 3
    )  # An Follow, Accept and Create activity should be there
    assert instance2_debug["outbox"] == 2  # We've sent a Accept and a Follow  activity

    # Ensure the post is visible in instance2's stream
    instance2_inbox_stream = instance2.stream_jsonfeed()
    assert len(instance2_inbox_stream["items"]) == 1
    assert instance2_inbox_stream["items"][0]["id"] == instance1_create_id

    instance2_create_id = instance2.new_note(
        f"hey @instance1@{instance1.docker_url}",
        reply=f"{instance1_create_id}/activity",
    )
    instance2_debug = instance2.debug()
    assert (
        instance2_debug["inbox"] == 3
    )  # An Follow, Accept and Create activity should be there
    assert (
        instance2_debug["outbox"] == 3
    )  # We've sent a Accept and a Follow and a Create  activity

    instance1_debug = instance1.debug()
    assert (
        instance1_debug["inbox"] == 3
    )  # An Follow, Accept and Create activity should be there
    assert (
        instance1_debug["outbox"] == 3
    )  # We've sent a Accept and a Follow and a Create  activity

    instance1_inbox_stream = instance1.stream_jsonfeed()
    assert len(instance1_inbox_stream["items"]) == 1
    assert instance1_inbox_stream["items"][0]["id"] == instance2_create_id

    instance1_note = instance1.outbox_get(f"{instance1_create_id}/activity")
    assert "replies" in instance1_note
    assert instance1_note["replies"]["totalItems"] == 1

    instance2.delete(f"{instance2_create_id}/activity")

    instance1_debug = instance1.debug()
    assert (
        instance1_debug["inbox"] == 4
    )  # An Follow, Accept and Create and Delete activity should be there
    assert (
        instance1_debug["outbox"] == 3
    )  # We've sent a Accept and a Follow and a Create  activity

    instance1_note = instance1.outbox_get(f"{instance1_create_id}/activity")
    assert "replies" in instance1_note
    assert instance1_note["replies"]["totalItems"] == 0


code:
username: 'instance1'
name: 'Instance 1'
icon_url: 'https://sos-ch-dk-2.exo.io/microblogpub/microblobpub.png'
domain: 'instance1_web:5005'
summary: 'instance1 summary'
pass: '$2b$12$nEgJMgaYbXSPOvgnqM4jSeYnleKhXqsFgv/o3hg12x79uEdsR4cUy'  # hello
https: false


code:
username: 'instance2'
name: 'Instance 2'
icon_url: 'https://sos-ch-dk-2.exo.io/microblogpub/microblobpub.png'
domain: 'instance2_web:5005'
summary: 'instance2 summary'
pass: '$2b$12$nEgJMgaYbXSPOvgnqM4jSeYnleKhXqsFgv/o3hg12x79uEdsR4cUy'  # hello
https: false


code:
username: 'ci'
name: 'CI tests'
icon_url: 'https://sos-ch-dk-2.exo.io/microblogpub/microblobpub.png'
domain: 'localhost:5005'
summary: 'test instance summary'
pass: '$2b$12$nEgJMgaYbXSPOvgnqM4jSeYnleKhXqsFgv/o3hg12x79uEdsR4cUy'  # hello
https: false


code:
import os

import pytest
import requests
from html2text import html2text

@pytest.fixture
def config():
    """Return the current config as a dict."""
    import yaml

    with open(
        os.path.join(os.path.dirname(__file__), "..", "config/me.yml"), "rb"
    ) as f:
        yield yaml.load(f)

def resp2plaintext(resp):
    """Convert the body of a requests reponse to plain text in order to make basic assertions."""
    return html2text(resp.text)

def test_ping_homepage(config):
    """Ensure the homepage is accessible."""
    resp = requests.get("http://localhost:5005")
    resp.raise_for_status()
    assert resp.status_code == 200
    body = resp2plaintext(resp)
    assert config["name"] in body
    assert f"@{config['username']}@{config['domain']}" in body


code:
import logging

logger = logging.getLogger(__name__)

def strtobool(s: str) -> bool:
    if s in ["y", "yes", "true", "on", "1"]:
        return True
    if s in ["n", "no", "false", "off", "0"]:
        return False

    raise ValueError(f"cannot convert {s} to bool")


code:
import binascii
import os
from typing import Callable

from little_boxes.key import Key

KEY_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "config")

def _new_key() -> str:
    return binascii.hexlify(os.urandom(32)).decode("utf-8")

def get_secret_key(name: str, new_key: Callable[[], str] = _new_key) -> str:
    """Loads or generates a cryptographic key."""
    key_path = os.path.join(KEY_DIR, f"{name}.key")
    if not os.path.exists(key_path):
        k = new_key()
        with open(key_path, "w+") as f:
            f.write(k)
        return k

    with open(key_path) as f:
        return f.read()

def get_key(owner: str, user: str, domain: str) -> Key:
    """"Loads or generates an RSA key."""
    k = Key(owner)
    user = user.replace(".", "_")
    domain = domain.replace(".", "_")
    key_path = os.path.join(KEY_DIR, f"key_{user}_{domain}.pem")
    if os.path.isfile(key_path):
        with open(key_path) as f:
            privkey_pem = f.read()
            k.load(privkey_pem)
    else:
        k.new()
        with open(key_path, "w") as f:
            f.write(k.privkey_pem)

    return k


code:
import json

import little_boxes.activitypub as ap
import mf2py
import requests
from little_boxes.errors import NotAnActivityError
from little_boxes.webfinger import get_actor_url

def lookup(url: str) -> ap.BaseActivity:
    """Try to find an AP object related to the given URL."""
    try:
        if url.startswith("@"):
            actor_url = get_actor_url(url)
            if actor_url:
                return ap.fetch_remote_activity(actor_url)
    except NotAnActivityError:
        pass
    except requests.HTTPError:
        # Some websites may returns 404, 503 or others when they don't support webfinger, and we're just taking a guess
        # when performing the lookup.
        pass

    backend = ap.get_backend()
    resp = requests.get(
        url,
        timeout=15,
        allow_redirects=False,
        headers={"User-Agent": backend.user_agent()},
    )
    resp.raise_for_status()

    # If the page is HTML, maybe it contains an alternate link pointing to an AP object
    for alternate in mf2py.parse(resp.text).get("alternates", []):
        if alternate.get("type") == "application/activity+json":
            return ap.fetch_remote_activity(alternate["url"])

    try:
        # Maybe the page was JSON-LD?
        data = resp.json()
        return ap.parse_activity(data)
    except json.JSONDecodeError:
        pass

    # Try content negotiation (retry with the AP Accept header)
    return ap.fetch_remote_activity(url)


code:
import base64
import mimetypes
from enum import Enum
from gzip import GzipFile
from io import BytesIO
from typing import Any

import gridfs
import piexif
import requests
from PIL import Image

def load(url, user_agent):
    """Initializes a `PIL.Image` from the URL."""
    with requests.get(url, stream=True, headers={"User-Agent": user_agent}) as resp:
        resp.raise_for_status()
        if not resp.headers.get("content-type").startswith("image/"):
            raise ValueError(f"bad content-type {resp.headers.get('content-type')}")

        resp.raw.decode_content = True
        return Image.open(BytesIO(resp.raw.read()))

def to_data_uri(img):
    out = BytesIO()
    img.save(out, format=img.format)
    out.seek(0)
    data = base64.b64encode(out.read()).decode("utf-8")
    return f"data:{img.get_format_mimetype()};base64,{data}"

class Kind(Enum):
    ATTACHMENT = "attachment"
    ACTOR_ICON = "actor_icon"
    UPLOAD = "upload"
    OG_IMAGE = "og"

class MediaCache(object):
    def __init__(self, gridfs_db: str, user_agent: str) -> None:
        self.fs = gridfs.GridFS(gridfs_db)
        self.user_agent = user_agent

    def cache_og_image(self, url: str) -> None:
        if self.fs.find_one({"url": url, "kind": Kind.OG_IMAGE.value}):
            return
        i = load(url, self.user_agent)
        # Save the original attachment (gzipped)
        i.thumbnail((100, 100))
        with BytesIO() as buf:
            with GzipFile(mode="wb", fileobj=buf) as f1:
                i.save(f1, format=i.format)
            buf.seek(0)
            self.fs.put(
                buf,
                url=url,
                size=100,
                content_type=i.get_format_mimetype(),
                kind=Kind.OG_IMAGE.value,
            )

    def cache_og_image2(self, url: str, remote_id: str) -> None:
        if self.fs.find_one({"url": url, "kind": Kind.OG_IMAGE.value}):
            return
        i = load(url, self.user_agent)
        # Save the original attachment (gzipped)
        i.thumbnail((100, 100))
        with BytesIO() as buf:
            with GzipFile(mode="wb", fileobj=buf) as f1:
                i.save(f1, format=i.format)
            buf.seek(0)
            self.fs.put(
                buf,
                url=url,
                size=100,
                content_type=i.get_format_mimetype(),
                kind=Kind.OG_IMAGE.value,
                remote_id=remote_id,
            )

    def cache_attachment(self, url: str) -> None:
        if self.fs.find_one({"url": url, "kind": Kind.ATTACHMENT.value}):
            return
        if (
            url.endswith(".png")
            or url.endswith(".jpg")
            or url.endswith(".jpeg")
            or url.endswith(".gif")
        ):
            i = load(url, self.user_agent)
            # Save the original attachment (gzipped)
            with BytesIO() as buf:
                f1 = GzipFile(mode="wb", fileobj=buf)
                i.save(f1, format=i.format)
                f1.close()
                buf.seek(0)
                self.fs.put(
                    buf,
                    url=url,
                    size=None,
                    content_type=i.get_format_mimetype(),
                    kind=Kind.ATTACHMENT.value,
                )
            # Save a thumbnail (gzipped)
            i.thumbnail((720, 720))
            with BytesIO() as buf:
                with GzipFile(mode="wb", fileobj=buf) as f1:
                    i.save(f1, format=i.format)
                buf.seek(0)
                self.fs.put(
                    buf,
                    url=url,
                    size=720,
                    content_type=i.get_format_mimetype(),
                    kind=Kind.ATTACHMENT.value,
                )
            return

    def cache_attachment2(self, url: str, remote_id: str) -> None:
        if self.fs.find_one({"url": url, "kind": Kind.ATTACHMENT.value}):
            return
        if (
            url.endswith(".png")
            or url.endswith(".jpg")
            or url.endswith(".jpeg")
            or url.endswith(".gif")
        ):
            i = load(url, self.user_agent)
            # Save the original attachment (gzipped)
            with BytesIO() as buf:
                f1 = GzipFile(mode="wb", fileobj=buf)
                i.save(f1, format=i.format)
                f1.close()
                buf.seek(0)
                self.fs.put(
                    buf,
                    url=url,
                    size=None,
                    content_type=i.get_format_mimetype(),
                    kind=Kind.ATTACHMENT.value,
                    remote_id=remote_id,
                )
            # Save a thumbnail (gzipped)
            i.thumbnail((720, 720))
            with BytesIO() as buf:
                with GzipFile(mode="wb", fileobj=buf) as f1:
                    i.save(f1, format=i.format)
                buf.seek(0)
                self.fs.put(
                    buf,
                    url=url,
                    size=720,
                    content_type=i.get_format_mimetype(),
                    kind=Kind.ATTACHMENT.value,
                    remote_id=remote_id,
                )
            return

        # The attachment is not an image, download and save it anyway
        with requests.get(
            url, stream=True, headers={"User-Agent": self.user_agent}
        ) as resp:
            resp.raise_for_status()
            with BytesIO() as buf:
                with GzipFile(mode="wb", fileobj=buf) as f1:
                    for chunk in resp.iter_content():
                        if chunk:
                            f1.write(chunk)
                buf.seek(0)
                self.fs.put(
                    buf,
                    url=url,
                    size=None,
                    content_type=mimetypes.guess_type(url)[0],
                    kind=Kind.ATTACHMENT.value,
                    remote_id=remote_id,
                )

    def cache_actor_icon(self, url: str) -> None:
        if self.fs.find_one({"url": url, "kind": Kind.ACTOR_ICON.value}):
            return
        i = load(url, self.user_agent)
        for size in [50, 80]:
            t1 = i.copy()
            t1.thumbnail((size, size))
            with BytesIO() as buf:
                with GzipFile(mode="wb", fileobj=buf) as f1:
                    t1.save(f1, format=i.format)
                buf.seek(0)
                self.fs.put(
                    buf,
                    url=url,
                    size=size,
                    content_type=i.get_format_mimetype(),
                    kind=Kind.ACTOR_ICON.value,
                )

    def save_upload(self, obuf: BytesIO, filename: str) -> str:
        # Remove EXIF metadata
        if filename.lower().endswith(".jpg") or filename.lower().endswith(".jpeg"):
            obuf.seek(0)
            with BytesIO() as buf2:
                piexif.remove(obuf.getvalue(), buf2)
                obuf.truncate(0)
                obuf.write(buf2.getvalue())

        obuf.seek(0)
        mtype = mimetypes.guess_type(filename)[0]
        with BytesIO() as gbuf:
            with GzipFile(mode="wb", fileobj=gbuf) as gzipfile:
                gzipfile.write(obuf.getvalue())

            gbuf.seek(0)
            oid = self.fs.put(
                gbuf,
                content_type=mtype,
                upload_filename=filename,
                kind=Kind.UPLOAD.value,
            )
            return str(oid)

    def cache(self, url: str, kind: Kind) -> None:
        if kind == Kind.ACTOR_ICON:
            self.cache_actor_icon(url)
        elif kind == Kind.OG_IMAGE:
            self.cache_og_image(url)
        else:
            self.cache_attachment(url)

    def get_actor_icon(self, url: str, size: int) -> Any:
        return self.get_file(url, size, Kind.ACTOR_ICON)

    def get_attachment(self, url: str, size: int) -> Any:
        return self.get_file(url, size, Kind.ATTACHMENT)

    def get_file(self, url: str, size: int, kind: Kind) -> Any:
        return self.fs.find_one({"url": url, "size": size, "kind": kind.value})


code:
import logging
import opengraph
import requests
from bs4 import BeautifulSoup
from little_boxes import activitypub as ap
from little_boxes.errors import NotAnActivityError
from little_boxes.urlutils import check_url
from little_boxes.urlutils import is_url_valid

from .lookup import lookup

logger = logging.getLogger(__name__)

def links_from_note(note):
    tags_href = set()
    for t in note.get("tag", []):
        h = t.get("href")
        if h:
            tags_href.add(h)

    links = set()
    if "content" in note:
        soup = BeautifulSoup(note["content"])
        for link in soup.find_all("a"):
            h = link.get("href")
            if h.startswith("http") and h not in tags_href and is_url_valid(h):
                links.add(h)

    # FIXME(tsileo): support summary and name fields

    return links

def fetch_og_metadata(user_agent, links):
    res = []
    for l in links:
        check_url(l)

        # Remove any AP actor from the list
        try:
            p = lookup(l)
            if p.has_type(ap.ACTOR_TYPES):
                continue
        except NotAnActivityError:
            pass

        r = requests.get(l, headers={"User-Agent": user_agent}, timeout=15)
        r.raise_for_status()
        if not r.headers.get("content-type").startswith("text/html"):
            logger.debug(f"skipping {l}")
            continue

        r.encoding = "UTF-8"
        html = r.text
        try:
            data = dict(opengraph.OpenGraph(html=html))
        except Exception:
            logger.exception(f"failed to parse {l}")
            continue
        if data.get("url"):
            res.append(data)

    return res


code:
.directory
.fuse*
*.pisi
*.pyc
*.svn
*~
.dir*
*.kate-swp

repo_name: ayhanyalcinsoy/main

path: .gitignore


code:
Herhangi bir paketin *-devel paketinin ba??ml?l???n?n tam olarak yaz?lmas?, pakete ba??ml? di?er paketlerinde d?zg?n bir ?ekilde derlenmesi sa?layacakt?r. Bu sebeple yap?lan, derlenen paketlerin, derlendikten sonra;

checkelf -s -x *.pisi

??kt?s?n?n haricinde bir yolu daha var ki buda, derlenme dizinine gidip (/var/pisi/*) buradaki install dizinimizde /usr/lib/pkgconfig/*.pc i?erisindeki *.pc dosyas?n?n i?eri?ini okumakt?r.

Bu dosyan?n i?erisinde;

Requires.private:
Requires:

yani ba??ml? ve ?zel ba??ml? olarak kar??s?nda ba??ml? paketler yazmaktad?r. burada yazanlar? pisi sf ile sistemde tart?p ne oldu?unu bulabiliriz.

?rnek;

libaccounts-glib  paketinin :  install dizini /var/pisi/libaccounts-glib-1.21-1/install

pkgconfig dizini : /var/pisi/libaccounts-glib-1.21-1/install/usr/lib/pkgconfig/libaccounts-glib.pc

libaccounts-glib.pc dosyas?n?n i?eri?i;

prefix=/usr
exec_prefix=${prefix}
libdir=${exec_prefix}/lib
includedir=${prefix}/include
applicationfilesdir=${prefix}/share/accounts/applications
servicefilesdir=${prefix}/share/accounts/services
servicetypefilesdir=${prefix}/share/accounts/service_types
providerfilesdir=${prefix}/share/accounts/providers

Name: libaccounts-glib
Description: A low-level library for managing accounts settings.
Requires: glib-2.0 gobject-2.0 gio-unix-2.0
Require.private: libxml-2.0 sqlite3
Version: 1.21
Libs: -L${libdir} -laccounts-glib
Cflags: -I${includedir}

konsolu a??p;

pisi sf glib-2.0 (?ok fazla sonu? ??kacakt?r ama i?lerinden bu glib-2.0 bulmak zor de?il ismiyle ayn? paket olacak)

glib2-devel paketi oldu?unu ??reneceksiniz.

-devel ba??ml?l?klar?n? bu ?ekilde bulabiliriz.


code:
#!/usr/bin/python
#
# -*- coding: utf-8 -*-
#
# Copyright (C) 2008, TUBITAK/UEKAE
#
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free
# Software Foundation; either version 2 of the License, or (at your option)
# any later version.
#
# Please read the COPYING file.
#

import sys
import multiprocessing

import urllib2
import bz2
import lzma

import piksemel

import pisi
import pisi.dependency as dependency
from pisi.graph import CycleException

class SourceDB:
    def __init__(self, index):

        self.__source_nodes = {}
        self.__pkgstosrc = {}

        doc = piksemel.parseString(index)
        self.__source_nodes, self.__pkgstosrc = self.__generate_sources(doc)

    def __generate_sources(self, doc):
        sources = {}
        pkgstosrc = {}

        for spec in doc.tags("SpecFile"):
            src_name = spec.getTag("Source").getTagData("Name")
            sources[src_name] = spec.toString()
            for package in spec.tags("Package"):
                pkgstosrc[package.getTagData("Name")] = src_name

        return sources, pkgstosrc

    def has_spec(self, name):
        return self.__source_nodes.has_key(name)

    def get_spec(self, name):
        src = self.__source_nodes[name]
        spec = pisi.specfile.SpecFile()
        spec.parse(src)
        return spec

    def list_specs(self):
        return self.__source_nodes.keys()

    def pkgtosrc(self, name):
        return self.__pkgstosrc[name]

def find_circle(sourcedb, A):

    G_f = pisi.graph.Digraph()

    def get_spec(name):
        if sourcedb.has_spec(name):
            return sourcedb.get_spec(name)
        else:
            raise Exception('Cannot find source package: %s' % name)

    def get_src(name):
        return get_spec(name).source

    def add_src(src):
        if not str(src.name) in G_f.vertices():
            G_f.add_vertex(str(src.name), (src.version, src.release))

    def pkgtosrc(pkg):
        try:
            tmp = sourcedb.pkgtosrc(pkg)
        except KeyError, e:
            # this is a bad hack but after we hit a problem we need to continue
            tmp = "e3"
            print "---> borks in ", e

        return tmp

    B = A

    install_list = set()

    while len(B) > 0:
        Bp = set()
        for x in B:
            sf = get_spec(x)
            src = sf.source
            add_src(src)

            # add dependencies

            def process_dep(dep):
                srcdep = pkgtosrc(dep.package)
                if not srcdep in G_f.vertices():
                    Bp.add(srcdep)
                    add_src(get_src(srcdep))
                if not src.name == srcdep: # firefox - firefox-devel thing
                    G_f.add_edge(src.name, srcdep)

            for builddep in src.buildDependencies:
                process_dep(builddep)

            for pkg in sf.packages:
                for rtdep in pkg.packageDependencies:
                    process_dep(rtdep)
        B = Bp

        try:
            order_build = G_f.topological_sort()
            order_build.reverse()
        except CycleException, cycle:
            return str(cycle)

    return ""

def getIndex(uri):
    try:
        if "://" in uri:
            rawdata = urllib2.urlopen(uri).read()
        else:
            rawdata = open(uri, "r").read()
    except IOError:
        print "could not fetch %s" % uri
        return None

    if uri.endswith("bz2"):
        data = bz2.decompress(rawdata)
    elif uri.endswith("xz") or uri.endswith("lzma"):
        data = lzma.decompress(rawdata)
    else:
        data = rawdata

    return data

def processPackage(pkg, sourcesLength, counter):
    global sourcedb

    sys.stdout.write("\r(%04d/%d) Calculating build dep of %s                       " % (counter, sourcesLength, pkg))
    sys.stdout.flush()

    return find_circle(sourcedb, [pkg])

def updateStatus(circleResult):
    global cycles
    cycles.add(circleResult)

if __name__ == "__main__":

    if len(sys.argv) < 2:
        print "Usage: circlefinder.py <source repo pisi-index.xml file>"
        sys.exit(1)

    rawIndex = getIndex(sys.argv[1])
    sourcedb = SourceDB(rawIndex)
    sources = sourcedb.list_specs()

    sourcesLength = len(sources)
    counter = 0

    global cycles
    cycles = set()

    pool = multiprocessing.Pool()

    for pkg in sources:
        counter += 1
        pool.apply_async(processPackage, (pkg, sourcesLength, counter), callback=updateStatus)

    pool.close()
    pool.join()

    if len(cycles):
        print
        for cycle in cycles:
            print cycle
    else:
        print "No circular dep found"

path: circledep-finder


code:
#!/usr/bin/python
# -*- coding: utf-8 -*-

import shutil
import sys
import re
import os

from optparse import OptionParser

def read_file(path):
    with open(path) as f:
        return f.read().strip()

def write_file(path, content):
    open(path, "w").write(content)
    open(path, "a").write("\n")

class Components():
    COMPONENT_XML = """<PISI>
    <Name>%s</Name>
</PISI>
"""
    EMPTY_COMPONENT = """        <Component>
            <Name>%s</Name>
            <LocalName xml:lang="en">FIXME</LocalName>
            FIXME
            <Description xml:lang="en">FIXME</Description>
            <Group>FIXME</Group>
            <Maintainer>
                <Name>PisiLinux Community</Name>
                <Email>admins@pisilinux.org</Email>
            </Maintainer>
        </Component>"""
    CBGN = """<PISI>
    <Components>"""
    CEND = """
    </Components>
</PISI>"""
    LOCALNAME = '            <LocalName xml:lang="%s">%s</LocalName>\n'
    SUMMARY = '            %s\n'
    DESCRIPTION = '            <Description xml:lang="%s">%s</Description>\n'
    COMPONENT = """
        <Component>
            <Name>%s</Name>
%s%s%s            <Group>%s</Group>
            <Maintainer>
                <Name>%s</Name>
                <Email>%s</Email>
            </Maintainer>
        </Component>"""

    def __init__(self, path):
        self.path = path
        self.components = []
        self.components_xml = "%s/components.xml" % self.path
        self.name_ptrn = re.compile("\s*<Name>(.+?)<\/Name>\s*")
        self.lnamech_ptrn = re.compile("\s*<LocalName.*?>(.+?)<\/LocalName>\s*")
        self.lname_ptrn = re.compile("\s*<LocalName\s+xml:lang\s?=\s?[\'\"](\w\w.*?)[\'\"]\s?>(.+?)<\/LocalName>\s*")
        self.summarych_ptrn = re.compile("\s*(.+?)<\/Summary>\s*")
        self.summary_ptrn = re.compile("\s*(.+?)<\/Summary>\s*")
        self.descch_ptrn = re.compile("\s*<Description.*?>(.+?)<\/Description>\s*")
        self.desc_ptrn = re.compile("\s*<Description\s+xml:lang\s?=\s?[\'\"](\w\w.*?)[\'\"]\s?>(.+?)<\/Description>\s*")
        self.group_ptrn = re.compile("\s*<Group>(.+?)<\/Group>\s*")
        self.email_ptrn = re.compile("\s*<Email>(.+?)<\/Email>\s*")
        self.cbgn_ptrn = re.compile("\s*<Component>\s*")
        self.cend_ptrn = re.compile("\s*<\/Component>\s*")
        self.mbgn_ptrn = re.compile("\s*<Maintainer>\s*")
        self.mend_ptrn = re.compile("\s*<\/Maintainer>\s*")
        self.csend_ptrn = re.compile("\s*<\/Components>\s*")

    def read_components(self):
        csfile = read_file(self.components_xml)
        cread = False
        mread = False
        component = {}
        for line in csfile.split("\n"):
            if re.search(self.cbgn_ptrn, line):
                cread = True
                component["err"] = []
                component["unknown"] = []
                component["clname"] = []
                component["csummary"] = []
                component["cdesc"] = []
                continue
            elif re.search(self.cend_ptrn, line):
                cread = False
                self.components.append(component)
                component = {}
                continue
            if cread:
                if re.search(self.mbgn_ptrn, line):
                    mread = True
                    continue
                elif re.search(self.mend_ptrn, line):
                    mread = False
                    continue
                elif re.search(self.name_ptrn, line):
                    if mread: component["mname"] = re.sub(self.name_ptrn, r"\1", line)
                    else: component["cname"] = re.sub(self.name_ptrn, r"\1", line)
                elif re.search(self.lnamech_ptrn, line):
                    if re.search(self.lname_ptrn, line):
                        component["clname"].append(re.sub(self.lname_ptrn, r"\1:\2", line))
                    else:
                        component["err"].append("LocalName")
                elif re.search(self.summarych_ptrn, line):
                    if re.search(self.summary_ptrn, line):
                        component["csummary"].append(re.sub(self.summary_ptrn, r"\1:\2", line))
                    else:
                        component["err"].append("Summary")
                elif re.search(self.descch_ptrn, line):
                    if re.search(self.desc_ptrn, line):
                        component["cdesc"].append(re.sub(self.desc_ptrn, r"\1:\2", line))
                    else:
                        component["err"].append("Description")
                elif re.search(self.group_ptrn, line):
                    component["cgroup"] = re.sub(self.group_ptrn, r"\1", line)
                elif re.search(self.email_ptrn, line):
                    component["memail"] = re.sub(self.email_ptrn, r"\1", line)
                else: component["unknown"].append(line)

    def sort_components(self):
        self.components = sorted(self.components, key=lambda k: k['cname'])

    def backup_components(self):
        files = os.walk(self.path).next()[2]
        last = 0
        for f in files:
            if f.startswith("components.xml.") and int(f[-3:]) > last: last = int(f[-3:])
        last = str(last + 1)
        while len(last) < 3: last = "0" + last
        shutil.copyfile(self.components_xml, ".".join((self.components_xml, last)))

    def write_components(self):
        self.sort_components()
        cs = ''
        for c in self.components:
            ln = ""
            for i in c["clname"]:
                ln += self.LOCALNAME % tuple(i.split(":"))
            s = ""
            for i in c["csummary"]:
                s += self.SUMMARY % tuple(i.split(":"))
            d = ""
            for i in c["cdesc"]:
                d += self.DESCRIPTION % tuple(i.split(":"))
            cs += self.COMPONENT % (c["cname"], ln, s, d, c["cgroup"], c["mname"], c["memail"])
        self.backup_components()
        open(self.components_xml, "w").write(self.CBGN)
        open(self.components_xml, "a").write(cs)
        open(self.components_xml, "a").write(self.CEND)
        open(self.components_xml, "a").write("\n")

    def edit(self):
        l = []
        mcommands = {"q": "quit",
                    "l": "list components",
                    "c": "choose component",
                    "h": "help"}
        ecommands = {"mn": "set maintainer name",
                     "me": "set maintainer email",
                     "ln": "set local name (lang:name)",
                     "s": "set summary (lang:summary)",
                     "d": "set description (lang:description)",
                     "m": "main menu",
                     "w": "write",
                     "h": "help"}

        def read_command(m):
            opts = mcommands.keys() if m == "m" else ecommands.keys()
            return (opts, raw_input(":".join(sorted(opts)) + " > ").split())

        def list_components(param, regexp = True):
            res = []
            num = 0
            if not param: param = ".*"
            for c in self.components:
                if regexp and re.search(param, c["cname"]):
                    res.append(c["cname"])
                    print num, c["cname"]
                    num += 1
                elif not regexp and param == c["cname"]:
                    res.append(c["cname"])
                    break
            return res

        def get_component_data(name):
            #print self.components
            for c in self.components:
                if name == c["cname"]: return c
            print "Component %s not found!" % name
            return False

        def print_help(hdict):
            for i in iter(sorted(hdict.iteritems())):
                print i[0] + "\t" + i[1]

        def edit_loop(component):
            def update(key, val, data):
                d = val.split(":")
                if not len(d) == 2: print "Wrong param format! (ln lang:text)"
                else:
                    lexists = -1
                    for i in data[key]:
                        if i.split(":")[0] == d[0]:
                            lexists = data[key].index(i)
                            break
                    if lexists > -1: data[key][lexists] = val
                    else: data[key].append(val)
                return data

            data = get_component_data(component)
            while True:
                print "-" * (len(component) + 8)
                print "Editing " + component
                print "-" * (len(component) + 8)
                print "LocalName:"
                for i in data["clname"]: print "  %s" %i
                print "Summary:"
                for i in data["csummary"]: print "  %s" %i
                print "Description:"
                for i in data["cdesc"]: print "  %s" %i
                print "Group:\n  %s" % data["cgroup"]
                print "Maintainer name:\n  %s" % data["mname"]
                print "Email:\n  %s" % data["memail"]
                o, c = read_command("c")
                if not c[0] in o: continue
                if len(c) == 1: c.append("")
                if c[0] == "m": break
                elif c[0] == "h": print_help(ecommands)
                elif c[0] == "ln": data = update("clname", c[1], data)
                elif c[0] == "s": data = update("csummary", c[1], data)
                elif c[0] == "d": data = update("cdesc", c[1], data)
                elif c[0] == "mn": data["mname"] = c[1]
                elif c[0] == "me": data["memail"] = c[1]
                elif c[0] == "w":  self.write_components()
                else: print"%s is not implemented yet" % c[0]

        def edit_component(param, l):
            try:
                int(param)
                edit_loop(l[int(param)])
            except ValueError:
                l = list_components(param, regexp = False)
            if not len(l) == 1: print "Component %s not found" % param
            else: edit_loop(l[0])

        while True:
            o, c = read_command("m")
            if len(c) == 1: c.append("")
            if not c[0] in o: continue
            elif c[0] == "q": break
            elif c[0] == "l": l = list_components(c[1])
            elif c[0] == "c": edit_component(c[1], l)
            elif c[0] == "h": print_help(mcommands)
            else: print"%s is not implemented yet" % c[0]

    def check(self):
        cs = []
        for root, dirs, files in os.walk(self.path):
            c = root.split(self.path)[1][1:].split("/")
            if "files" in c or "comar" in c: continue
            c = ".".join(c)
            component_xml = "%s/component.xml" % root
            pspec_xml = "%s/pspec.xml" % root
            actions_py = "%s/actions.py" % root
            if not os.path.isfile(component_xml) and not os.path.isfile(pspec_xml):
                if os.path.isfile(actions_py): print "WARNING: %s not exists" % pspec_xml
                else:
                    is_src_repo = False
                    for r, d, f in os.walk(root):
                        if "pspec.xml" in f:
                            is_src_repo = True
                            break
                    if is_src_repo and c:
                        print "%s not exists. creating..." % component_xml
                        write_file(component_xml, self.COMPONENT_XML % c)
            if os.path.isfile(component_xml) and c: cs.append(c)

        mcs = cs[:]
        csfile = read_file(self.components_xml)
        maintainer = False
        new_file = []
        for line in csfile.split("\n"):
            new_file.append(line)
            if re.search(self.mbgn_ptrn, line): maintainer = True
            elif re.search(self.mend_ptrn, line): maintainer = False
            elif re.search(self.csend_ptrn, line):
                for m in mcs:
                    new_file.insert(-1, self.EMPTY_COMPONENT % m)
            if not re.search(self.name_ptrn, line) or maintainer: continue
            cn = re.sub(self.name_ptrn, r"\1", line)
            if cn in mcs: mcs.pop(mcs.index(cn))

        new_file = "\n".join(new_file)
        write_file(self.components_xml, new_file)

if __name__ == "__main__":
    usage = "Usage: %prog [PATH]"
    parser = OptionParser(usage)
    parser.add_option("-c", "--check", action="store_true", dest="check", help="fix missing component.xml files and entries in components.xml")
    parser.add_option("-e", "--edit", action="store_true", dest="edit", help="edit components.xml")
    (options,args) = parser.parse_args()
    try:
        root = args[0]
    except IndexError:
        print "Using ./ as PATH"
        root = "./"

    if root.endswith("/"): root = root[:-1]

    cs = Components(root)
    if not os.path.isfile(cs.components_xml):
        print "%s not exists!" % cs.components_xml
        sys.exit(1)
    if not os.access(cs.components_xml, os.W_OK):
        print "Cannot write to %s" % cs.components_xml
        sys.exit(1)
    cs.read_components()

    if options.check: cs.check()
    if options.edit:
        cs.edit()