# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
-import argparse
-import atoma
-import bs4
import csv
import inspect
-import itertools
import json
-import re
-import reqto
import sys
import time
+
+import argparse
+import atoma
+import bs4
+import markdown
import validators
from fba import blacklist
from fba import config
from fba import fba
from fba import instances
+from fba import network
from fba.federation import *
# DEBUG: print(f"DEBUG: args[]={type(args)} - CALLED!")
domains = list()
try:
- fetched = fba.post_json_api("gql.api.bka.li", "/v1/graphql", json.dumps({
+ fetched = network.post_json_api("gql.api.bka.li", "/v1/graphql", json.dumps({
"query": "query domainlist {nodeinfo(order_by: {domain: asc}) {domain}}"
}))
for entry in fetched["data"]["nodeinfo"]:
# DEBUG: print(f"DEBUG: entry['{type(entry)}']='{entry}'")
if not "domain" in entry:
- print(f"WARNING: entry does not contain 'domain' - SKIPPED!")
+ print(f"WARNING: entry()={len(entry)} does not contain 'domain' - SKIPPED!")
continue
elif not validators.domain(entry["domain"]):
print(f"WARNING: domain='{entry['domain']}' is not a valid domain - SKIPPED!")
# DEBUG: print(f"DEBUG: Adding domain='{entry['domain']}' ...")
domains.append(entry["domain"])
- except BaseException as e:
- print(f"ERROR: Cannot fetch graphql,exception[{type(e)}]:'{str(e)}'")
+ except BaseException as exception:
+ print(f"ERROR: Cannot fetch graphql,exception[{type(exception)}]:'{str(exception)}'")
sys.exit(255)
# DEBUG: print(f"DEBUG: domains()={len(domains)}")
def fetch_blocks(args: argparse.Namespace):
# DEBUG: print(f"DEBUG: args[]={type(args)} - CALLED!")
- if args.domain != None and args.domain != "":
+ if args.domain is not None and args.domain != "":
# DEBUG: print(f"DEBUG: args.domain='{args.domain}' - checking ...")
if not validators.domain(args.domain):
print(f"WARNING: domain='{args.domain}' is not valid.")
boot.acquire_lock()
- if args.domain != None and args.domain != "":
+ if args.domain is not None and args.domain != "":
# Re-check single domain
fba.cursor.execute(
"SELECT domain, software, origin, nodeinfo_url FROM instances WHERE software IN ('pleroma', 'mastodon', 'friendica', 'misskey', 'bookwyrm', 'takahe') AND domain = ?", [args.domain]
print(f"INFO: blocker='{blocker}',software='{software}'")
try:
if software == "friendica":
- json = fba.fetch_friendica_blocks(blocker)
+ rows = friendica.fetch_blocks(blocker)
elif software == "misskey":
- json = misskey.fetch_blocks(blocker)
+ rows = misskey.fetch_blocks(blocker)
- print(f"INFO: Checking {len(json.items())} entries from blocker='{blocker}',software='{software}' ...")
- for block_level, blocklist in json.items():
+ print(f"INFO: Checking {len(rows.items())} entries from blocker='{blocker}',software='{software}' ...")
+ for block_level, blocklist in rows.items():
# DEBUG: print("DEBUG: blocker,block_level,blocklist():", blocker, block_level, len(blocklist))
block_level = fba.tidyup_domain(block_level)
# DEBUG: print("DEBUG: AFTER-block_level:", block_level)
blocked, reason = block.values()
# DEBUG: print(f"DEBUG: blocked='{blocked}',reason='{reason}' - BEFORE!")
blocked = fba.tidyup_domain(blocked)
- reason = fba.tidyup_reason(reason) if reason != None and reason != "" else None
+ reason = fba.tidyup_reason(reason) if reason is not None and reason != "" else None
# DEBUG: print(f"DEBUG: blocked='{blocked}',reason='{reason}' - AFTER!")
if blocked == "":
searchres = fba.cursor.fetchone()
- if searchres == None:
+ if searchres is None:
print(f"WARNING: Cannot deobsfucate blocked='{blocked}' - SKIPPED!")
continue
searchres = fba.cursor.fetchone()
- if searchres == None:
+ if searchres is None:
print(f"WARNING: Cannot deobsfucate blocked='{blocked}' - SKIPPED!")
continue
# DEBUG: print("DEBUG: Committing changes ...")
fba.connection.commit()
- except Exception as e:
- print(f"ERROR: blocker='{blocker}',software='{software}',exception[{type(e)}]:'{str(e)}'")
+ except BaseException as exception:
+ print(f"ERROR: blocker='{blocker}',software='{software}',exception[{type(exception)}]:'{str(exception)}'")
else:
print("WARNING: Unknown software:", blocker, software)
if config.get("bot_enabled") and len(blockdict) > 0:
- send_bot_post(blocker, blockdict)
+ network.send_bot_post(blocker, blockdict)
blockdict = []
def fetch_cs(args: argparse.Namespace):
# DEBUG: print(f"DEBUG: args[]={type(args)} - CALLED!")
+ extensions = [
+ 'extra',
+ 'abbr',
+ 'attr_list',
+ 'def_list',
+ 'fenced_code',
+ 'footnotes',
+ 'md_in_html',
+ 'admonition',
+ 'codehilite',
+ 'legacy_attrs',
+ 'legacy_em',
+ 'meta',
+ 'nl2br',
+ 'sane_lists',
+ 'smarty',
+ 'toc',
+ 'wikilinks'
+ ]
+
domains = {
"silenced": list(),
"reject" : list(),
}
try:
- doc = bs4.BeautifulSoup(
- fba.get_response("meta.chaos.social", "/federation", fba.headers, (config.get("connection_timeout"), config.get("read_timeout"))).text,
- "html.parser",
- )
- # DEBUG: print(f"DEBUG: doc()={len(doc)}[]={type(doc)}")
- silenced = doc.find("h2", {"id": "silenced-instances"}).findNext("table")
+ raw = fba.fetch_url("https://raw.githubusercontent.com/chaossocial/meta/master/federation.md", network.web_headers, (config.get("connection_timeout"), config.get("read_timeout"))).text
+ # DEBUG: print(f"DEBUG: raw()={len(raw)}[]={type(raw)}")
+
+ doc = bs4.BeautifulSoup(markdown.markdown(raw, extensions=extensions), features='html.parser')
+ # DEBUG: print(f"DEBUG: doc()={len(doc)}[]={type(doc)}")
+ silenced = doc.find("h2", {"id": "silenced-instances"}).findNext("table").find("tbody")
# DEBUG: print(f"DEBUG: silenced[]={type(silenced)}")
domains["silenced"] = domains["silenced"] + fba.find_domains(silenced)
- blocked = doc.find("h2", {"id": "blocked-instances"}).findNext("table")
+ blocked = doc.find("h2", {"id": "blocked-instances"}).findNext("table").find("tbody")
# DEBUG: print(f"DEBUG: blocked[]={type(blocked)}")
domains["reject"] = domains["reject"] + fba.find_domains(blocked)
- except BaseException as e:
- print(f"ERROR: Cannot fetch from meta.chaos.social,exception[{type(e)}]:'{str(e)}'")
+ except BaseException as exception:
+ print(f"ERROR: Cannot fetch from meta.chaos.social,exception[{type(exception)}]:'{str(exception)}'")
sys.exit(255)
# DEBUG: print(f"DEBUG: domains()={len(domains)}")
try:
print(f"INFO: Fetch FBA-specific RSS args.feed='{args.feed}' ...")
- response = fba.fetch_url(args.feed, fba.headers, (config.get("connection_timeout"), config.get("read_timeout")))
+ response = fba.fetch_url(args.feed, network.web_headers, (config.get("connection_timeout"), config.get("read_timeout")))
# DEBUG: print(f"DEBUG: response.ok={response.ok},response.status_code='{response.status_code}',response.text()={len(response.text)}")
if response.ok and response.status_code < 300 and len(response.text) > 0:
# DEBUG: print(f"DEBUG: Adding domain='{domain}'")
domains.append(domain)
- except BaseException as e:
- print(f"ERROR: Cannot fetch feed='{feed}',exception[{type(e)}]:'{str(e)}'")
+ except BaseException as exception:
+ print(f"ERROR: Cannot fetch feed='{feed}',exception[{type(exception)}]:'{str(exception)}'")
sys.exit(255)
# DEBUG: print(f"DEBUG: domains()={len(domains)}")
domains = list()
try:
print(f"INFO: Fetching ATOM feed='{feed}' from FBA bot account ...")
- response = fba.fetch_url(feed, fba.headers, (config.get("connection_timeout"), config.get("read_timeout")))
+ response = fba.fetch_url(feed, network.web_headers, (config.get("connection_timeout"), config.get("read_timeout")))
# DEBUG: print(f"DEBUG: response.ok={response.ok},response.status_code='{response.status_code}',response.text()={len(response.text)}")
if response.ok and response.status_code < 300 and len(response.text) > 0:
# DEBUG: print(f"DEBUG: Adding domain='{domain}',domains()={len(domains)}")
domains.append(domain)
- except BaseException as e:
- print(f"ERROR: Cannot fetch feed='{feed}',exception[{type(e)}]:'{str(e)}'")
+ except BaseException as exception:
+ print(f"ERROR: Cannot fetch feed='{feed}',exception[{type(exception)}]:'{str(exception)}'")
sys.exit(255)
# DEBUG: print(f"DEBUG: domains({len(domains)})={domains}")
boot.acquire_lock()
# Fetch this URL
- response = fba.fetch_url("https://github.com/federater/blocks_recommended/raw/main/federater.csv", fba.headers, (config.get("connection_timeout"), config.get("read_timeout")))
+ response = fba.fetch_url("https://github.com/federater/blocks_recommended/raw/main/federater.csv", network.web_headers, (config.get("connection_timeout"), config.get("read_timeout")))
# DEBUG: print(f"DEBUG: response[]='{type(response)}'")
if response.ok and response.content != "":
# DEBUG: print(f"DEBUG: Fetched {len(response.content)} Bytes, parsing CSV ...")
projects / fba.git / blobdiff