tarekziade

from flask import make_response
from urllib.parse import urlparse

# domain:port
SAFE_DOMAINS = ['ziade.org:443']


@app.after_request
def check_redirect(response):
    if response.status_code != 302:

tarekziade

import asyncio
from smwogger import API


async def run_heartbeat():
    async with API('__api__.json') as api:
        res = await api.__heartbeat__()
        body = await res.json()
        print(body)

tarekziade

import asyncio
import configparser
import pytest
from smwogger import API


@pytest.fixture
def conf():
    config = configparser.ConfigParser()
    config.read('manifest.ini')

tarekziade

# see https://dxr.mozilla.org/mozilla-central/source/testing/marionette/driver.js?from=testing%2Fmarionette%2Fdriver.js#2957
import functools
import json
from asyncio import open_connection
import asyncio
from molotov import *


class Marionette(object):
    def __init__(self, host='localhost', port=2828, loop=None):

tarekziade

import functools
import logging
import graypy
import json
import time
import random
from collections import defaultdict, deque
from flask import Flask, jsonify, g

app = Flask(__name__)

tarekziade

/* Like for the IntersectionObserver, we're using here the
 * Mutation observer to detect when an element attribute is changed
 * by some Javascript on the page.
 *
 * use case: a button is enabled
 */
function watchMutation(selector, condition) {
  var target = document.querySelector(selector);
  var observer = new MutationObserver(function(mutations) {
    mutations.forEach(function(mutation) {

tarekziade

class ConcurrentRunner:
    def __init__(self, max_concurrency=5):
        self.max_concurrency = 5
        self.tasks = []

    async def put(self, coro):
        """Starts a coroutine if there are 4 or less already running ones.
        """
        # blocks until there's room
        while len(self.tasks) >= self.max_concurrency:

tarekziade

# taken from https://stackoverflow.com/a/37211337
def _make_sleep():
    async def sleep(delay, result=None, *, loop=None):
        coro = asyncio.sleep(delay, result=result)
        task = asyncio.ensure_future(coro, loop=loop)
        sleep.tasks.add(task)
        try:
            return await task
        except asyncio.CancelledError:
            return result

tarekziade

"""
Demo of extracting text and links from a rendered web page.

$ brew install geckodriver
$ python3 -m venv .
$ bin/pip install bs4 selenium
$ bin/python scrap.py

The script looks for an element of a specific id on the page.
This can be used to make sure we wait for all JS to execute, and

tarekziade

import sys
import nltk
from nltk.corpus import wordnet as wn

nltk.download("wordnet")


def find_hypernyms(synset_name):
    synset = wn.synset(synset_name)
    hypernyms = synset.hypernyms()