Date 문자열(RFC822)을 UTC Timestamp로 변경

Python 2019.04.23 23:26 Posted by 김한별 behonestar
rfc822_date = 'Wed, 21 Oct 2015 07:28:00 GMT'
timestamp = email.utils.mktime_tz((email.utils.parsedate_tz(rfc822_date))) --> 1445412480

 

[python] 문자열 n번째 자리에 문자 삽입하기

Python 2018.06.21 13:11 Posted by 김한별 behonestar

소스코드

s = "00115faabbcc"


s2 = re.sub(r'(.{2})', r':\1', s)[1:]


print(s2)


출력

00:11:5f:aa:bb:cc


윈도우에서 python mysqlclient 패키지 설치하기

Python 2017.12.22 16:05 Posted by 김한별 behonestar

1. pip용 whl 파일 다운로드 


mysqlclient-1.3.12-cp27-cp27m-win32.whl



2. 설치


pip install mysqlclient-1.3.12-cp27-cp27m-win32.whl

[Python] sqlalchemy "Mysql Gone Away" 오류

Python 2017.05.22 16:04 Posted by 김한별 behonestar

증상

sqlalchemy로 DB 세션을 열었더라도 mysql의 wait_timeout동안 요청이 없으면 mysql에서 연결을 끊어버립니다. 따라서 이후의 요청은 Mysql Gone Away 오류가 발생합니다.


대책

mysql의 wait_timeout보다 작은 값으로 pool_cycle을 지정하면 됩니다. 특정 시간 간격으로 재연결하는 옵션입니다.


engine = create_engine("mysql://%s:%s@%s:%s/mydb" % (db_user, db_pass, db_host, db_port), pool_recycle=3600, echo=False)


AWS Cognito Python Sample Code

Python 2017.01.11 11:09 Posted by 김한별 behonestar

AWS Cognito를 통해 임시 AWS credentials를 발급받는 Python 예제 코드입니다.

ADMIN_NO_SRP_AUTH 방식을 사용하기 때문에 AWS credentials가 제공된 환경에서 동작합니다. 


Pre-Requirements

  • AWS Cognito의 User Pool과 Identity Pool이 생성되어 있어야합니다.
  • User Pool App을 생성할 때 Generate client secret 옵션이 비활성화되어야 합니다.
  • User Pool App의 ADMIN_NO_SRP_AUTH 옵션이 활성화되어야 합니다.

Auth Flow



# -*- coding: utf-8 -*-

import boto3


class Cognito():

    region = 'ap-northeast-2'

    user_pool_id = 'ap-northeast-2_xxxxxxxxx'

    app_client_id = 'hjohd8ijicoxxxxxxxxxxxxxx'

    identity_pool_id = 'ap-northeast-2:xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx'

    account_id = 'xxxxxxxxxxxx'


    def sign_up(self, username, password, email):

        idp_client = boto3.client('cognito-idp')

        resp = idp_client.sign_up(ClientId=self.app_client_id,

                                  Username=username,

                                  Password=password,

                                  UserAttributes=[{'Name': 'email', 'Value': email}])

        return resp


    def confirm_sign_up(self, username, confirm_code):

        idp_client = boto3.client('cognito-idp')

        resp = idp_client.confirm_sign_up(ClientId=self.app_client_id,

                                          Username=username,

                                          ConfirmationCode=confirm_code)

        return resp


    def sign_in_admin(self, username, password):

        # Get ID Token

        idp_client = boto3.client('cognito-idp')

        resp = idp_client.admin_initiate_auth(UserPoolId=self.user_pool_id,

                                              ClientId=self.app_client_id,

                                              AuthFlow='ADMIN_NO_SRP_AUTH',

                                              AuthParameters={'USERNAME': username,'PASSWORD': password})

        

        provider = 'cognito-idp.%s.amazonaws.com/%s' % (self.region, self.user_pool_id)

        token = resp['AuthenticationResult']['IdToken']


        # Get IdentityId

        ci_client = boto3.client('cognito-identity')

        resp = ci_client.get_id(AccountId=self.account_id,

                               IdentityPoolId=self.identity_pool_id,

                               Logins={provider:token})

        

        # Get Credentials

        resp = ci_client.get_credentials_for_identity(IdentityId=resp['IdentityId'],

                                                      Logins={provider: token})

        return resp



[python] 트래픽 초과 시 메일 발송하기

Python 2016.12.01 11:24 Posted by 김한별 behonestar

트래픽 한계치, GMail 계정과 비밀번호를 설정해야 한다.


#!/usr/bin/env python from email.MIMEMultipart import MIMEMultipart from email.MIMEBase import MIMEBase from email.MIMEText import MIMEText from email.header import Header from email import Encoders from urllib2 import urlopen import time import math import smtplib MAIL_SEND_INTERVAL = 3600 #sec INTERVAL = 1 #sec WARN_KBPS = 212234 #kbps PUB_IP = urlopen('http://ip.42.pl/raw').read()


ignore = 0 prev_bytes = -1 rate_history = [] def GetNetworkInterfaces(): f = open("/proc/net/dev") data = f.read() f.close() data = data.split("\n")[2:] for i in data: i = i.replace(':', ' ') if len(i.strip()) > 0: x = i.split() k = { "interface" : x[0][:len( x[0])-1], "tx" : { "bytes" : int(x[1]), "packets" : int(x[2]), "errs" : int(x[3]), "drop" : int(x[4]), "fifo" : int(x[5]), "frame" : int(x[6]), "compressed" : int(x[7]), "multicast" : int(x[8]) }, "rx" : { "bytes" : int(x[9]), "packets" : int(x[10]), "errs" : int(x[11]), "drop" : int(x[12]), "fifo" : int(x[13]), } } if k["interface"] == "eth": return k return None def send_email(to, bitrate): gmail_username="yourname" gmail_user="user@yourdomain.com" gmail_pwd="***********" msg=MIMEMultipart('alternative') msg['From']=gmail_username msg['To']=to msg['Subject']=Header("RELAY SERVER [%s] TRAFFIC WARNING!" % PUB_IP,'utf-8') msg.attach(MIMEText("Average Rates: %d kbps" % bitrate, 'plain', 'utf-8')) mailServer=smtplib.SMTP("smtp.gmail.com",587) mailServer.ehlo() mailServer.starttls() mailServer.ehlo() mailServer.login(gmail_user,gmail_pwd) mailServer.sendmail(gmail_user, to, msg.as_string()) mailServer.close() print "send email [%s]" % to while True: idata = GetNetworkInterfaces() if prev_bytes == -1: prev_bytes = idata["rx"]["bytes"] continue cur_rate = (idata["rx"]["bytes"] - prev_bytes) / INTERVAL prev_bytes = idata["rx"]["bytes"] if len(rate_history) > 10: rate_history.pop(0) rate_history.append(cur_rate) sum_rate = 0 for rate in rate_history: sum_rate += rate avg_rate = sum_rate / len(rate_history) avg_rate = avg_rate*8/1024 print "CUR: %d kbps AVG: %d kpbs" % (cur_rate*8/1024, avg_rate) if avg_rate > WARN_KBPS: ignore -= 1 if ignore < 0: send_email("user@yourdomain.com", avg_rate) ignore = MAIL_SEND_INTERVAL #1hour  

    time.sleep(INTERVAL)


python 스크립트 데몬으로 실행하기

Python 2016.05.09 16:23 Posted by 김한별 behonestar

1. 데몬 스크립트 작성

# sudo su

# cd /etc/init.d

# touch tunnel

# chmod 755 ./tunnel

# vi ./tunnel


nohup 사용

- 세션이 종료되어도 백그라운드로 프로세스 동작

- 로그를 파일로 출력

#!/bin/bash

# chkconfig: 345 88 08

# description: Tunneling Server Deamon


VENV=/home/ec2-user/venv

WORKDIR=/home/ec2-user/tunnel

DAEMON=tunnel.py

LOG=/var/log/tunnel.log


function do_start()

{

        source ${VENV}/bin/activate

        cd ${WORKDIR}

        nohup python ${DAEMON} & >> ${LOG}

}


function do_stop()

{

        PID=`ps -ef | grep ${DAEMON} | grep -v grep | awk '{print $2}'`

        if [ "$PID" != "" ]; then

                kill -9 $PID

        fi

}


case "$1" in

    start|stop)

        do_${1}

        ;;

    reload|restart)

        do_stop

        do_start

        ;;

    *)

        echo "Usage: /etc/init.d/tunnel {start|stop|restart}"

        exit 1

        ;;

esac



2. 부팅시 자동실행 등록

# chkconfig --add tunnel

# chkconfig --list | grep tunnel


TAG Python, SCRIPT

Redis를 DB Table처럼 활용하기

Python 2016.05.04 17:29 Posted by 김한별 behonestar

데이터 저장 및 탐색 방법


test.py

import redis


r = redis.StrictRedis(host=host, port=port, db=0)


# input

r.hmset('id:a', {'ip':'192.168.10.1', 'port':1000})

r.hmset('id:b', {'ip':'192.168.10.2', 'port':1001})

r.hmset('id:c', {'ip':'192.168.10.3', 'port':1002})

r.hmset('id:d', {'ip':'192.168.10.4', 'port':1003})


# output

keys = r.keys('id:*')

for key in keys:

    item = r.hgetall(key)

    print '%s:%s' % (item['ip'], item['port'])


결과

$ python test.py


192.168.10.2:1001

192.168.10.4:1003

192.168.10.3:1002

192.168.10.1:1000



데이터 저장 및 쿼리 방법



TAG redis