martes, 21 de agosto de 2012

Twitter, Python y GNU/Linux

Hola, bueno el otro dice haciendo algunas travesuras :P deje encendida mi maquina procesando algunas cosas, pero por ciertos motivos queria lograr cancelar esta tarea remotamente, pero el problema es que estaba en el cine (primer inconveniente), el segundo inconveniente fue que no tengo SSH Server público en mi PC por el hecho que no tengo IP pública (segundo inconveniente) y finalmente uno diría utilizar algo como TeamViewer ya que hay para dispositivos Android, esta solución parecería la más razonable pero yo no cuento con un smartphone (tercer inconveniente). Y bueno, se me vino una solución que igual realice hace un tiempo, pero utilizando CSharp bajo Windows.

 La idea de este proyecto o del que hice en windows es el de escribir un tweet y que la aplicación lea constantemente el último tweet de cierto usuario. Ahora si el tweet tiene cierto formato preestablecido por nosotros indicando que sea cierta tarea se debe realizar en la PC que ejecute esta aplicación. 

Pues bien, en windows lo logre sin muchos problemas utilizando Twitterizer. Ahora quiero hacer algo similar para que se ejecute en GNU/Linux, para ello utilizaré el buen Python (nadie le gana XDD). 


Investigando un poco sobre la API de Twitter me di cuenta que para obtener los tweets de cierto usuario no es necesario contar con un access_token o cosas similares, simplemente se necesita la siguiente URL con los parámetros que estan en la documentación oficial. 

 https://api.twitter.com/1/statuses/user_timeline.json 

 En este caso utilice los siguientes parámetros:
  • screen_name: nombre del usuario del cual se desea obtener los tweets
  • count: indica el número de tweets que se obtendrán
  • trim_user: en pocas palabras para simplificar la salida JSON en la parte de información de usuario
Bueno, ahora entrando un poco a la parte de python, simplemente se tiene que enviar una Request a esta URL utilizando los parámetros que se mencionaron, para ello se utiliza el módulo urllib2. Realice una clase que abstraiga el proceso de obtener los últimos 9 tweets y que permita obtener el último tweet de un usuario si es que existe (este método se llama a petición). El código para realizar este proceso quedo como sigue:

# Name: twitter.py
# Desc: Clase que abstrae la logica de obtener un tweet
# Coded by Donkeysharp
import urllib2
import json

API_URL = 'https://api.twitter.com/1/statuses/user_timeline.json?screen_name=%s&count=%d&trim_user=%s'
NO_TWEET = 'NT'

class Twitter:
   def __init__(self, username, trim_user = True, tweets_count = 10):
      # Opciones API_TWITTER
      self.user = username
      self.trim_user = str(trim_user)
      self.tweets_count = tweets_count
      self.tweets = []
      
      self.__load_tweets()

   def __build_url(self):
      return API_URL % (self.user, self.tweets_count, self.trim_user)

   """ 
      Envia una peticion a Twitter para obtener los tweets 
      del usuario especificado
   """
   def __get_twitter_response(self, tweets_count = 10):
      try:
         # Lee el ultimo tweet de un usuario
         response = urllib2.urlopen(self.__build_url())
         json_response = response.read()
         # Realiza el parsin de cadena JSON a estructuras list/dict de Python
         json_list = json.loads(json_response)
         return json_list
      except Exception, e:
         # Posiblemente ERROR de conexion
         print e
         return []
         
         
   def __parse_json_entry(self, entry):
      return {'id':entry['id_str'], 'text':entry['text']}
      
   """
      Convierte la estructura de twitter a una mas simple {id, text}
   """
   def __load_tweets(self):
      json_list = self.__get_twitter_response()
      for item in json_list:
         self.tweets.append(self.__parse_json_entry(item))
         
   """
      Envia una peticion a Twitter para obtener solamente
      el ultimo tweeet
   """
   def get_last_tweet(self):
      json_list = self.__get_twitter_response(tweets_count=1)
      if json_list[0]['id_str'] != self.tweets[0]['id']:
         self.tweets.insert(0, self.__parse_json_entry(json_list[0]))
         
   def get_tweet(self, idx = 0):
      if 0 <= idx and idx < len(self.tweets):
         return (self.tweets[idx]['id'], self.tweets[idx]['text'])
      return ('-1', NO_TWEET)
   
Y la aplicación cliente como tal lo que hace es en un bucle infinito revisar por el último tweet y si este no es el mismo del último que ya tenemos, lo adiciona a la lista de tweets y verifica el texto para ver si es un comando con el formato que este preestablecido por nosotros. El código del cliente me quedo así:

#!/usr/bin/python
# Name: myzombie.py
# Desc: Script to control pc from twitter
# Coded by Donkeysharp
# 
# Comandos aceptados:
# zombie> calc
# zombie> shutdown
# zombie> 
import os
import subprocess
from twitter import *
import time

def nuevo_tweet(t, last_tweet):
   if t.tweets[0]['id'] == last_tweet:
      return False
   return True

def test_zombie_command(cmd):
   cmd = cmd.lower()
   if len(cmd) > 10:
      if cmd[0:11] == 'zombie> ':
         cmd = cmd[11:]
         print cmd
         if cmd == 'un_cmd':
            subprocess.Popen('un_cmd')
         elif cmd == 'otro_cmd':
            subprocess.Popen('otro_cmd')
   
def main():
   print '[-] Conectando con Twitter ...'
   t = Twitter(username = 'donkeysharp')
   # id,text = t.get_tweet()
   # t.get_last_tweet()
   print '       Obteniendo ultimos 9 tweets ...'
   
   print '[+] Ultimos 9 Tweets:'
   print '---------------------'
   for x in t.tweets:
      print '    ', x['id']
      time.sleep(0.1)
      
   if len(t.tweets) <= 0:
      return
      
   
   last_tweet = t.tweets[0]['id']
   try:
      print '[-] Esperando nuevos tweets ...'
      while True:
         # Obtendra el ultimo Tweet
         t.get_last_tweet()
         if nuevo_tweet(t, last_tweet):
            print '[+] Tweet encontrado:'
            print '---------------------'
            print '    ',t.tweets[0]['id']
            last_tweet = t.tweets[0]['id']
            
            test_zombie_command(t.tweets[0]['text'])
         
         # Realiza la verificacion de tweets cada 5 segundos
         time.sleep(5)
   except:
      pass

if __name__ == '__main__':
   main()

Como ven, todo radica en obtener los tweets y el que hacer con esos tweets ya depende de nosotros, apagar el ordenar, matar un proceso con cierto PID, reiniciar, etc. Esta me parece una alternativa interesante para controlar a ciegas una máquina remotamente. 

Y bueno, si funciono ya que lo probe desde mi celular que no cuenta ni con una camara y si funcionó. Bueno, una entrada un poco rápida pero espero les sea de utilidad. 

 Saludos!!

3 comentarios: