<s> import socket <EOL> from traceback import print_exc <EOL> from subnetparse import IP_List <EOL> from clock import clock <EOL> from __init__ import createPeerID <EOL> try : <EOL> True <EOL> except : <EOL> True = <NUM_LIT:1> <EOL> False = <NUM_LIT:0> <EOL> DEBUG = False <EOL> EXPIRE_CACHE = <NUM_LIT:30> <EOL> ID = "<STR_LIT>" + createPeerID ( ) [ - <NUM_LIT:4> : ] <EOL> try : <EOL> import pythoncom , win32com . client <EOL> _supported = <NUM_LIT:1> <EOL> except ImportError : <EOL> _supported = <NUM_LIT:0> <EOL> class _UPnP1 : <EOL> def __init__ ( self ) : <EOL> self . map = None <EOL> self . last_got_map = - <NUM_LIT> <EOL> def _get_map ( self ) : <EOL> if self . last_got_map + EXPIRE_CACHE < clock ( ) : <EOL> try : <EOL> dispatcher = win32com . client . Dispatch ( "<STR_LIT>" ) <EOL> self . map = dispatcher . StaticPortMappingCollection <EOL> self . last_got_map = clock ( ) <EOL> except : <EOL> self . map = None <EOL> return self . map <EOL> def test ( self ) : <EOL> try : <EOL> assert self . _get_map ( ) <EOL> success = True <EOL> except : <EOL> success = False <EOL> return success <EOL> def open ( self , ip , p ) : <EOL> map = self . _get_map ( ) <EOL> try : <EOL> map . Add ( p , '<STR_LIT>' , p , ip , True , ID ) <EOL> if DEBUG : <EOL> print '<STR_LIT>' + ip + '<STR_LIT::>' + str ( p ) <EOL> success = True <EOL> except : <EOL> if DEBUG : <EOL> print "<STR_LIT>" + str ( p ) <EOL> print_exc ( ) <EOL> success = False <EOL> return success <EOL> def close ( self , p ) : <EOL> map = self . _get_map ( ) <EOL> try : <EOL> map . Remove ( p , '<STR_LIT>' ) <EOL> success = True <EOL> if DEBUG : <EOL> print '<STR_LIT>' + str ( p ) <EOL> except : <EOL> if DEBUG : <EOL> print '<STR_LIT>' + str ( p ) <EOL> print_exc ( ) <EOL> success = False <EOL> return success <EOL> def clean ( self , retry = False ) : <EOL> if not _supported : <EOL> return <EOL> try : <EOL> map = self . _get_map ( ) <EOL> ports_in_use = [ ] <EOL> for i in xrange ( len ( map ) ) : <EOL> try : <EOL> mapping = map [ i ] <EOL> port = mapping . ExternalPort <EOL> prot = str ( mapping . Protocol ) . lower ( ) <EOL> desc = str ( mapping . Description ) . lower ( ) <EOL> except : <EOL> port = None <EOL> if port and prot == '<STR_LIT>' and desc [ : <NUM_LIT:3> ] == '<STR_LIT>' : <EOL> ports_in_use . append ( port ) <EOL> success = True <EOL> for port in ports_in_use : <EOL> try : <EOL> map . Remove ( port , '<STR_LIT>' ) <EOL> except : <EOL> success = False <EOL> if not success and not retry : <EOL> self . clean ( retry = True ) <EOL> except : <EOL> pass <EOL> class _UPnP2 : <EOL> def __init__ ( self ) : <EOL> self . services = None <EOL> self . last_got_services = - <NUM_LIT> <EOL> def _get_services ( self ) : <EOL> if not self . services or self . last_got_services + EXPIRE_CACHE < clock ( ) : <EOL> self . services = [ ] <EOL> try : <EOL> f = win32com . client . Dispatch ( "<STR_LIT>" ) <EOL> for t in ( "<STR_LIT>" , <EOL> "<STR_LIT>" ) : <EOL> try : <EOL> conns = f . FindByType ( t , <NUM_LIT:0> ) <EOL> for c in xrange ( len ( conns ) ) : <EOL> try : <EOL> svcs = conns [ c ] . Services <EOL> for s in xrange ( len ( svcs ) ) : <EOL> try : <EOL> self . services . append ( svcs [ s ] ) <EOL> except : <EOL> pass <EOL> except : <EOL> pass <EOL> except : <EOL> pass <EOL> except : <EOL> pass <EOL> self . last_got_services = clock ( ) <EOL> return self . services <EOL> def test ( self ) : <EOL> try : <EOL> assert self . _get_services ( ) <EOL> success = True <EOL> except : <EOL> success = False <EOL> return success <EOL> def open ( self , ip , p ) : <EOL> svcs = self . _get_services ( ) <EOL> success = False <EOL> for s in svcs : <EOL> try : <EOL> s . InvokeAction ( '<STR_LIT>' , [ '<STR_LIT>' , p , '<STR_LIT>' , p , ip , True , ID , <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> success = True <EOL> except : <EOL> pass <EOL> if DEBUG and not success : <EOL> print "<STR_LIT>" + str ( p ) <EOL> print_exc ( ) <EOL> return success <EOL> def close ( self , p ) : <EOL> svcs = self . _get_services ( ) <EOL> success = False <EOL> for s in svcs : <EOL> try : <EOL> s . InvokeAction ( '<STR_LIT>' , [ '<STR_LIT>' , p , '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> success = True <EOL> except : <EOL> pass <EOL> if DEBUG and not success : <EOL> print "<STR_LIT>" + str ( p ) <EOL> print_exc ( ) <EOL> return success <EOL> class _UPnP : <EOL> def __init__ ( self ) : <EOL> self . upnp1 = _UPnP1 ( ) <EOL> self . upnp2 = _UPnP2 ( ) <EOL> self . upnplist = ( None , self . upnp1 , self . upnp2 ) <EOL> self . upnp = None <EOL> self . local_ip = None <EOL> self . last_got_ip = - <NUM_LIT> <EOL> def get_ip ( self ) : <EOL> if self . last_got_ip + EXPIRE_CACHE < clock ( ) : <EOL> local_ips = IP_List ( ) <EOL> local_ips . set_intranet_addresses ( ) <EOL> try : <EOL> for info in socket . getaddrinfo ( socket . gethostname ( ) , <NUM_LIT:0> , socket . AF_INET ) : <EOL> self . local_ip = info [ <NUM_LIT:4> ] [ <NUM_LIT:0> ] <EOL> if local_ips . includes ( self . local_ip ) : <EOL> self . last_got_ip = clock ( ) <EOL> if DEBUG : <EOL> print '<STR_LIT>' + self . local_ip <EOL> break <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> except : <EOL> self . local_ip = None <EOL> if DEBUG : <EOL> print '<STR_LIT>' <EOL> print_exc ( ) <EOL> return self . local_ip <EOL> def test ( self , upnp_type ) : <EOL> if DEBUG : <EOL> print '<STR_LIT>' + str ( upnp_type ) <EOL> if not upnp_type or not _supported or self . get_ip ( ) is None : <EOL> if DEBUG : <EOL> print '<STR_LIT>' <EOL> return <NUM_LIT:0> <EOL> pythoncom . CoInitialize ( ) <EOL> self . upnp = self . upnplist [ upnp_type ] <EOL> if self . upnp . test ( ) : <EOL> if DEBUG : <EOL> print '<STR_LIT>' <EOL> return upnp_type <EOL> if DEBUG : <EOL> print '<STR_LIT>' <EOL> return <NUM_LIT:0> <EOL> def open ( self , p ) : <EOL> assert self . upnp , "<STR_LIT>" <EOL> return self . upnp . open ( self . get_ip ( ) , p ) <EOL> def close ( self , p ) : <EOL> assert self . upnp , "<STR_LIT>" <EOL> return self . upnp . close ( p ) <EOL> def clean ( self ) : <EOL> return self . upnp1 . clean ( ) <EOL> _upnp_ = _UPnP ( ) <EOL> UPnP_test = _upnp_ . test <EOL> UPnP_open_port = _upnp_ . open <EOL> UPnP_close_port = _upnp_ . close <EOL> UPnP_reset = _upnp_ . clean </s>
