jerryan999

#!/usr/bin/python3
# -*- coding: utf-8 -*-

class Agent(object):
  """ Specify single proxy agent object
      Atttribute:
          proxy: like "http://45.78.34.180:8080"
          success: this proxy's life value (just like solder's blood value in game),\
                  it minus one if failed and plus one if successed
          percentage: proxy's percentage of successful useage, successful_times/total_using-times,default 100%

jerryan999

class ProxyMiddleware(object):
  """ Customized Proxy Middleware
      No matter success or fail, change proxy for every request 
  """

  # Change another proxy instead of passing to RetryMiddlewares when met these errors
  DONT_RETRY_ERRORS = (TimeoutError,ConnectionRefusedError,TCPTimedOutError,
                      ResponseNeverReceived, ConnectError, ConnectBindError, TunnelError)

  agent_list = []

jerryan999

##
# Create a figure space matrix consisting of 3 columns and 2 rows
#
# Here is a useful template to use for working with subplots.
#
##################################################################
fig, ax = plt.subplots(figsize=(10,5), ncols=3, nrows=2)

left   =  0.125  # the left side of the subplots of the figure
right  =  0.9    # the right side of the subplots of the figure

jerryan999

def splitDataFrameList(df,target_column):
    ''' df = dataframe to split,
    target_column = the column containing the values to split
    returns: a dataframe with each entry for the target column separated, with each element moved into a new row. 
    The values in the other columns are duplicated across the newly divided rows.
    '''
    def splitListToRows(row,row_accumulator,target_column):
        split_row = row[target_column]
        for s in split_row:
            new_row = row.to_dict()

jerryan999

from multiprocessing import cpu_count, Process, Queue, Manager
import threading
import os
try:
  from Queue import Empty
except:
  from queue import Empty
from threading import Lock as TL
import threading
from multiprocessing import Lock as PL

jerryan999

import os
from sshtunnel import SSHTunnelForwarder

def connect_tunnel():
    local_port = 4599
    os.system("kill $( lsof -i:{} -t )".format(local_port))
    with SSHTunnelForwarder(
        ('34.230.115.45', 22),       # Remote server ip and port
        ssh_username="jerry",
        ssh_pkey="/Users/jerry/.ssh/id_ali.rsa",

jerryan999

from datetime import datetime, timedelta, timezone

TZ_DELTA = 0                            # Define timezone, UTC '0'
tzinfo = timezone(timedelta(hours=TZ_DELTA))

#  timezone-aware datetime
now = datetime.now(tz=tzinfo)

# normal datetime
now = datetime.now()

jerryan999

import time
import threading

from functools import wraps


def rate_limited(max_per_second):
    """
    Decorator that make functions not be called faster than
    """

jerryan999

import pickle
import argparse
from os import listdir
from os.path import isfile, join
import sys
def convert(input_file_with_path, output_file_with_path):
  with open(input_file_with_path, 'rb') as f_in:
    data = pickle.load(f_in)
    with open(output_file_with_path, 'wb') as f_out:
        pickle.dump(data, f_out, protocol=2)

jerryan999

"""
Qxf2: Example script to run one test against the Chess Free app using Appium
The test will:
- launch the app
- click the 'PLAY!' button
"""
 
import os
import unittest
from appium import webdriver